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, bool two_phase)
 
static void DecodeAbort (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
 
static void DecodePrepare (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
 
static void DecodeXLogTuple (char *data, Size len, ReorderBufferTupleBuf *tup)
 
static bool FilterPrepare (LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
 
static bool DecodeTXNNeedSkip (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid dbId, RepOriginId origin_id)
 
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,
bool  two_phase 
)
static

Definition at line 847 of file decode.c.

References xl_xact_parsed_abort::dbId, DecodeTXNNeedSkip(), XLogRecordBuffer::endptr, XLogReaderState::EndRecPtr, i, InvalidXLogRecPtr, xl_xact_parsed_abort::nsubxacts, xl_xact_parsed_abort::origin_lsn, xl_xact_parsed_abort::origin_timestamp, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAbort(), ReorderBufferFinishPrepared(), xl_xact_parsed_abort::subxacts, xl_xact_parsed_abort::twophase_gid, UpdateDecodingStats(), xl_xact_parsed_abort::xact_time, XACT_XINFO_HAS_ORIGIN, xl_xact_parsed_abort::xinfo, and XLogRecGetOrigin.

Referenced by DecodeXactOp().

850 {
851  int i;
852  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
853  TimestampTz abort_time = parsed->xact_time;
854  XLogRecPtr origin_id = XLogRecGetOrigin(buf->record);
855  bool skip_xact;
856 
857  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
858  {
859  origin_lsn = parsed->origin_lsn;
860  abort_time = parsed->origin_timestamp;
861  }
862 
863  /*
864  * Check whether we need to process this transaction. See
865  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
866  * transaction.
867  */
868  skip_xact = DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id);
869 
870  /*
871  * Send the final rollback record for a prepared transaction unless we
872  * need to skip it. For non-two-phase xacts, simply forget the xact.
873  */
874  if (two_phase && !skip_xact)
875  {
876  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
877  abort_time, origin_id, origin_lsn,
879  parsed->twophase_gid, false);
880  }
881  else
882  {
883  for (i = 0; i < parsed->nsubxacts; i++)
884  {
885  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
886  buf->record->EndRecPtr);
887  }
888 
889  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
890  }
891 
892  /* update the decoding stats */
893  UpdateDecodingStats(ctx);
894 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:172
struct ReorderBuffer * reorder
Definition: logical.h:42
int64 TimestampTz
Definition: timestamp.h:39
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
XLogRecPtr EndRecPtr
Definition: xlogreader.h:176
XLogRecPtr endptr
Definition: decode.c:49
XLogReaderState * record
Definition: decode.c:50
TransactionId * subxacts
Definition: xact.h:360
static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid dbId, RepOriginId origin_id)
Definition: decode.c:1308
void UpdateDecodingStats(LogicalDecodingContext *ctx)
Definition: logical.c:1803
XLogRecPtr origin_lsn
Definition: xact.h:368
static bool two_phase
uint64 XLogRecPtr
Definition: xlogdefs.h:21
TimestampTz xact_time
Definition: xact.h:353
void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, XLogRecPtr two_phase_at, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn, char *gid, bool is_commit)
XLogRecPtr origptr
Definition: decode.c:48
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
int i
TimestampTz origin_timestamp
Definition: xact.h:369
char twophase_gid[GIDSIZE]
Definition: xact.h:366

◆ DecodeCommit()

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

Definition at line 677 of file decode.c.

References xl_xact_parsed_commit::dbId, DecodeTXNNeedSkip(), XLogRecordBuffer::endptr, i, 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(), ReorderBufferFinishPrepared(), ReorderBufferForget(), SnapBuildCommitTxn(), SnapBuildGetTwoPhaseAt(), LogicalDecodingContext::snapshot_builder, xl_xact_parsed_commit::subxacts, xl_xact_parsed_commit::twophase_gid, UpdateDecodingStats(), xl_xact_parsed_commit::xact_time, XACT_XINFO_HAS_ORIGIN, xl_xact_parsed_commit::xinfo, and XLogRecGetOrigin.

Referenced by DecodeXactOp().

680 {
681  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
682  TimestampTz commit_time = parsed->xact_time;
683  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
684  int i;
685 
686  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
687  {
688  origin_lsn = parsed->origin_lsn;
689  commit_time = parsed->origin_timestamp;
690  }
691 
693  parsed->nsubxacts, parsed->subxacts);
694 
695  /* ----
696  * Check whether we are interested in this specific transaction, and tell
697  * the reorderbuffer to forget the content of the (sub-)transactions
698  * if not.
699  *
700  * We can't just use ReorderBufferAbort() here, because we need to execute
701  * the transaction's invalidations. This currently won't be needed if
702  * we're just skipping over the transaction because currently we only do
703  * so during startup, to get to the first transaction the client needs. As
704  * we have reset the catalog caches before starting to read WAL, and we
705  * haven't yet touched any catalogs, there can't be anything to invalidate.
706  * But if we're "forgetting" this commit because it happened in another
707  * database, the invalidations might be important, because they could be
708  * for shared catalogs and we might have loaded data into the relevant
709  * syscaches.
710  * ---
711  */
712  if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
713  {
714  for (i = 0; i < parsed->nsubxacts; i++)
715  {
716  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
717  }
718  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
719 
720  return;
721  }
722 
723  /* tell the reorderbuffer about the surviving subtransactions */
724  for (i = 0; i < parsed->nsubxacts; i++)
725  {
726  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
727  buf->origptr, buf->endptr);
728  }
729 
730  /*
731  * Send the final commit record if the transaction data is already
732  * decoded, otherwise, process the entire transaction.
733  */
734  if (two_phase)
735  {
736  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
738  commit_time, origin_id, origin_lsn,
739  parsed->twophase_gid, true);
740  }
741  else
742  {
743  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
744  commit_time, origin_id, origin_lsn);
745  }
746 
747  /*
748  * Update the decoding stats at transaction prepare/commit/abort.
749  * Additionally we send the stats when we spill or stream the changes to
750  * avoid losing them in case the decoding is interrupted. It is not clear
751  * that sending more or less frequently than this would be better.
752  */
753  UpdateDecodingStats(ctx);
754 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:172
struct ReorderBuffer * reorder
Definition: logical.h:42
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
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:346
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:937
int64 TimestampTz
Definition: timestamp.h:39
TransactionId * subxacts
Definition: xact.h:332
uint16 RepOriginId
Definition: xlogdefs.h:65
XLogRecPtr origin_lsn
Definition: xact.h:345
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
XLogRecPtr endptr
Definition: decode.c:49
XLogReaderState * record
Definition: decode.c:50
static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid dbId, RepOriginId origin_id)
Definition: decode.c:1308
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
void UpdateDecodingStats(LogicalDecodingContext *ctx)
Definition: logical.c:1803
static bool two_phase
struct SnapBuild * snapshot_builder
Definition: logical.h:43
uint64 XLogRecPtr
Definition: xlogdefs.h:21
void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, XLogRecPtr two_phase_at, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn, char *gid, bool is_commit)
XLogRecPtr SnapBuildGetTwoPhaseAt(SnapBuild *builder)
Definition: snapbuild.c:376
XLogRecPtr origptr
Definition: decode.c:48
int i
char twophase_gid[GIDSIZE]
Definition: xact.h:341
TimestampTz xact_time
Definition: xact.h:325

◆ DecodeDelete()

static void DecodeDelete ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

1029 {
1030  XLogReaderState *r = buf->record;
1031  xl_heap_delete *xlrec;
1032  ReorderBufferChange *change;
1033  RelFileNode target_node;
1034 
1035  xlrec = (xl_heap_delete *) XLogRecGetData(r);
1036 
1037  /* only interested in our database */
1038  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
1039  if (target_node.dbNode != ctx->slot->data.database)
1040  return;
1041 
1042  /* output plugin doesn't look for this origin, no need to queue */
1043  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1044  return;
1045 
1046  change = ReorderBufferGetChange(ctx->reorder);
1047 
1048  if (xlrec->flags & XLH_DELETE_IS_SUPER)
1050  else
1052 
1053  change->origin_id = XLogRecGetOrigin(r);
1054 
1055  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1056 
1057  /* old primary key stored */
1058  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
1059  {
1060  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
1061  Size tuplelen = datalen - SizeOfHeapHeader;
1062 
1063  Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
1064 
1065  change->data.tp.oldtuple =
1066  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1067 
1068  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
1069  datalen, change->data.tp.oldtuple);
1070  }
1071 
1072  change->data.tp.clear_toast_afterwards = true;
1073 
1075  change, false);
1076 }
#define XLH_DELETE_IS_SUPER
Definition: heapam_xlog.h:99
struct ReorderBuffer * reorder
Definition: logical.h:42
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1262
#define XLH_DELETE_CONTAINS_OLD
Definition: heapam_xlog.h:103
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:147
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:610
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:85
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:321
#define SizeOfHeapDelete
Definition: heapam_xlog.h:115
XLogReaderState * record
Definition: decode.c:50
union ReorderBufferChange::@97 data
RepOriginId origin_id
Definition: reorderbuffer.h:90
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1531
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
#define Assert(condition)
Definition: c.h:804
size_t Size
Definition: c.h:540
struct ReorderBufferChange::@97::@98 tp
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define SizeOfHeapHeader
Definition: heapam_xlog.h:151
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeHeap2Op()

static void DecodeHeap2Op ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 440 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_FREEZE_PAGE, XLOG_HEAP2_LOCK_UPDATED, XLOG_HEAP2_MULTI_INSERT, XLOG_HEAP2_NEW_CID, XLOG_HEAP2_PRUNE, XLOG_HEAP2_REWRITE, XLOG_HEAP2_VACUUM, XLOG_HEAP2_VISIBLE, XLOG_HEAP_OPMASK, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

Referenced by LogicalDecodingProcessRecord().

441 {
443  TransactionId xid = XLogRecGetXid(buf->record);
444  SnapBuild *builder = ctx->snapshot_builder;
445 
446  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
447 
448  /*
449  * If we don't have snapshot or we are just fast-forwarding, there is no
450  * point in decoding changes.
451  */
453  ctx->fast_forward)
454  return;
455 
456  switch (info)
457  {
459  if (!ctx->fast_forward &&
460  SnapBuildProcessChange(builder, xid, buf->origptr))
461  DecodeMultiInsert(ctx, buf);
462  break;
463  case XLOG_HEAP2_NEW_CID:
464  {
465  xl_heap_new_cid *xlrec;
466 
467  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
468  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
469 
470  break;
471  }
472  case XLOG_HEAP2_REWRITE:
473 
474  /*
475  * Although these records only exist to serve the needs of logical
476  * decoding, all the work happens as part of crash or archive
477  * recovery, so we don't need to do anything here.
478  */
479  break;
480 
481  /*
482  * Everything else here is just low level physical stuff we're not
483  * interested in.
484  */
486  case XLOG_HEAP2_PRUNE:
487  case XLOG_HEAP2_VACUUM:
488  case XLOG_HEAP2_VISIBLE:
490  break;
491  default:
492  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
493  }
494 }
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1120
struct ReorderBuffer * reorder
Definition: logical.h:42
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:59
uint32 TransactionId
Definition: c.h:587
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:53
unsigned char uint8
Definition: c.h:439
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
#define XLOG_HEAP2_PRUNE
Definition: heapam_xlog.h:54
#define XLOG_HEAP2_MULTI_INSERT
Definition: heapam_xlog.h:58
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:722
#define XLOG_HEAP2_VISIBLE
Definition: heapam_xlog.h:57
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
#define ERROR
Definition: elog.h:46
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:772
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:60
XLogReaderState * record
Definition: decode.c:50
#define XLOG_HEAP2_VACUUM
Definition: heapam_xlog.h:55
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:315
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLOG_HEAP2_FREEZE_PAGE
Definition: heapam_xlog.h:56
XLogRecPtr origptr
Definition: decode.c:48
#define elog(elevel,...)
Definition: elog.h:232
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)

◆ DecodeHeapOp()

static void DecodeHeapOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

501 {
503  TransactionId xid = XLogRecGetXid(buf->record);
504  SnapBuild *builder = ctx->snapshot_builder;
505 
506  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
507 
508  /*
509  * If we don't have snapshot or we are just fast-forwarding, there is no
510  * point in decoding data changes.
511  */
513  ctx->fast_forward)
514  return;
515 
516  switch (info)
517  {
518  case XLOG_HEAP_INSERT:
519  if (SnapBuildProcessChange(builder, xid, buf->origptr))
520  DecodeInsert(ctx, buf);
521  break;
522 
523  /*
524  * Treat HOT update as normal updates. There is no useful
525  * information in the fact that we could make it a HOT update
526  * locally and the WAL layout is compatible.
527  */
529  case XLOG_HEAP_UPDATE:
530  if (SnapBuildProcessChange(builder, xid, buf->origptr))
531  DecodeUpdate(ctx, buf);
532  break;
533 
534  case XLOG_HEAP_DELETE:
535  if (SnapBuildProcessChange(builder, xid, buf->origptr))
536  DecodeDelete(ctx, buf);
537  break;
538 
539  case XLOG_HEAP_TRUNCATE:
540  if (SnapBuildProcessChange(builder, xid, buf->origptr))
541  DecodeTruncate(ctx, buf);
542  break;
543 
544  case XLOG_HEAP_INPLACE:
545 
546  /*
547  * Inplace updates are only ever performed on catalog tuples and
548  * can, per definition, not change tuple visibility. Since we
549  * don't decode catalog tuples, we're not interested in the
550  * record's contents.
551  *
552  * In-place updates can be used either by XID-bearing transactions
553  * (e.g. in CREATE INDEX CONCURRENTLY) or by XID-less
554  * transactions (e.g. VACUUM). In the former case, the commit
555  * record will include cache invalidations, so we mark the
556  * transaction as catalog modifying here. Currently that's
557  * redundant because the commit will do that as well, but once we
558  * support decoding in-progress relations, this will be important.
559  */
560  if (!TransactionIdIsValid(xid))
561  break;
562 
563  SnapBuildProcessChange(builder, xid, buf->origptr);
565  break;
566 
567  case XLOG_HEAP_CONFIRM:
568  if (SnapBuildProcessChange(builder, xid, buf->origptr))
569  DecodeSpecConfirm(ctx, buf);
570  break;
571 
572  case XLOG_HEAP_LOCK:
573  /* we don't care about row level locks for now */
574  break;
575 
576  default:
577  elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
578  break;
579  }
580 }
#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:902
uint32 TransactionId
Definition: c.h:587
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
unsigned char uint8
Definition: c.h:439
#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:961
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:722
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1226
#define ERROR
Definition: elog.h:46
static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1028
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:315
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
#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:1082
#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:232
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 902 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().

903 {
904  Size datalen;
905  char *tupledata;
906  Size tuplelen;
907  XLogReaderState *r = buf->record;
908  xl_heap_insert *xlrec;
909  ReorderBufferChange *change;
910  RelFileNode target_node;
911 
912  xlrec = (xl_heap_insert *) XLogRecGetData(r);
913 
914  /*
915  * Ignore insert records without new tuples (this does happen when
916  * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
917  */
918  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
919  return;
920 
921  /* only interested in our database */
922  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
923  if (target_node.dbNode != ctx->slot->data.database)
924  return;
925 
926  /* output plugin doesn't look for this origin, no need to queue */
927  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
928  return;
929 
930  change = ReorderBufferGetChange(ctx->reorder);
931  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
933  else
935  change->origin_id = XLogRecGetOrigin(r);
936 
937  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
938 
939  tupledata = XLogRecGetBlockData(r, 0, &datalen);
940  tuplelen = datalen - SizeOfHeapHeader;
941 
942  change->data.tp.newtuple =
943  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
944 
945  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
946 
947  change->data.tp.clear_toast_afterwards = true;
948 
950  change,
952 }
struct ReorderBuffer * reorder
Definition: logical.h:42
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1262
#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:147
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:610
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:85
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
#define XLH_INSERT_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:69
XLogReaderState * record
Definition: decode.c:50
union ReorderBufferChange::@97 data
RepOriginId origin_id
Definition: reorderbuffer.h:90
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1531
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1555
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
size_t Size
Definition: c.h:540
struct ReorderBufferChange::@97::@98 tp
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define SizeOfHeapHeader
Definition: heapam_xlog.h:151
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeLogicalMsgOp()

static void DecodeLogicalMsgOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

623 {
624  SnapBuild *builder = ctx->snapshot_builder;
625  XLogReaderState *r = buf->record;
626  TransactionId xid = XLogRecGetXid(r);
627  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
628  RepOriginId origin_id = XLogRecGetOrigin(r);
629  Snapshot snapshot;
630  xl_logical_message *message;
631 
632  if (info != XLOG_LOGICAL_MESSAGE)
633  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
634 
636 
637  /*
638  * If we don't have snapshot or we are just fast-forwarding, there is no
639  * point in decoding messages.
640  */
642  ctx->fast_forward)
643  return;
644 
645  message = (xl_logical_message *) XLogRecGetData(r);
646 
647  if (message->dbId != ctx->slot->data.database ||
648  FilterByOrigin(ctx, origin_id))
649  return;
650 
651  if (message->transactional &&
652  !SnapBuildProcessChange(builder, xid, buf->origptr))
653  return;
654  else if (!message->transactional &&
656  SnapBuildXactNeedsSkip(builder, buf->origptr)))
657  return;
658 
659  snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
660  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
661  message->transactional,
662  message->message, /* first part of message is
663  * prefix */
664  message->message_size,
665  message->message + message->prefix_size);
666 }
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:587
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:394
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
unsigned char uint8
Definition: c.h:439
uint16 RepOriginId
Definition: xlogdefs.h:65
ReplicationSlotPersistentData data
Definition: slot.h:147
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:610
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
char message[FLEXIBLE_ARRAY_MEMBER]
Definition: message.h:27
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:722
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
#define ERROR
Definition: elog.h:46
XLogRecPtr endptr
Definition: decode.c:49
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:315
bool transactional
Definition: message.h:23
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
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:232
#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:662

◆ DecodeMultiInsert()

static void DecodeMultiInsert ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

1121 {
1122  XLogReaderState *r = buf->record;
1123  xl_heap_multi_insert *xlrec;
1124  int i;
1125  char *data;
1126  char *tupledata;
1127  Size tuplelen;
1128  RelFileNode rnode;
1129 
1130  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
1131 
1132  /*
1133  * Ignore insert records without new tuples. This happens when a
1134  * multi_insert is done on a catalog or on a non-persistent relation.
1135  */
1136  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
1137  return;
1138 
1139  /* only interested in our database */
1140  XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
1141  if (rnode.dbNode != ctx->slot->data.database)
1142  return;
1143 
1144  /* output plugin doesn't look for this origin, no need to queue */
1145  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1146  return;
1147 
1148  /*
1149  * We know that this multi_insert isn't for a catalog, so the block should
1150  * always have data even if a full-page write of it is taken.
1151  */
1152  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
1153  Assert(tupledata != NULL);
1154 
1155  data = tupledata;
1156  for (i = 0; i < xlrec->ntuples; i++)
1157  {
1158  ReorderBufferChange *change;
1159  xl_multi_insert_tuple *xlhdr;
1160  int datalen;
1161  ReorderBufferTupleBuf *tuple;
1163 
1164  change = ReorderBufferGetChange(ctx->reorder);
1166  change->origin_id = XLogRecGetOrigin(r);
1167 
1168  memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
1169 
1170  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
1171  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
1172  datalen = xlhdr->datalen;
1173 
1174  change->data.tp.newtuple =
1175  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
1176 
1177  tuple = change->data.tp.newtuple;
1178  header = tuple->tuple.t_data;
1179 
1180  /* not a disk based tuple */
1182 
1183  /*
1184  * We can only figure this out after reassembling the transactions.
1185  */
1186  tuple->tuple.t_tableOid = InvalidOid;
1187 
1188  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1189 
1190  memset(header, 0, SizeofHeapTupleHeader);
1191 
1192  memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
1193  (char *) data,
1194  datalen);
1195  header->t_infomask = xlhdr->t_infomask;
1196  header->t_infomask2 = xlhdr->t_infomask2;
1197  header->t_hoff = xlhdr->t_hoff;
1198 
1199  /*
1200  * Reset toast reassembly state only after the last row in the last
1201  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
1202  * call.
1203  */
1204  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
1205  (i + 1) == xlrec->ntuples)
1206  change->data.tp.clear_toast_afterwards = true;
1207  else
1208  change->data.tp.clear_toast_afterwards = false;
1209 
1211  buf->origptr, change, false);
1212 
1213  /* move to the next xl_multi_insert_tuple entry */
1214  data += datalen;
1215  }
1216  Assert(data == tupledata + tuplelen);
1217 }
#define SizeofHeapTupleHeader
Definition: htup_details.h:184
struct ReorderBuffer * reorder
Definition: logical.h:42
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:147
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:610
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:85
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
HeapTupleHeader t_data
Definition: htup.h:68
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
#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
union ReorderBufferChange::@97 data
RepOriginId origin_id
Definition: reorderbuffer.h:90
#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:1531
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1555
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
#define InvalidOid
Definition: postgres_ext.h:36
#define Assert(condition)
Definition: c.h:804
size_t Size
Definition: c.h:540
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:212
struct ReorderBufferChange::@97::@98 tp
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:753
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:193
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodePrepare()

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

Definition at line 772 of file decode.c.

References xl_xact_parsed_commit::dbId, DecodeTXNNeedSkip(), XLogRecordBuffer::endptr, i, xl_xact_parsed_commit::nsubxacts, xl_xact_parsed_commit::origin_lsn, xl_xact_parsed_commit::origin_timestamp, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferCommitChild(), ReorderBufferInvalidate(), ReorderBufferPrepare(), ReorderBufferRememberPrepareInfo(), ReorderBufferSkipPrepare(), SNAPBUILD_CONSISTENT, SnapBuildCurrentState(), LogicalDecodingContext::snapshot_builder, xl_xact_parsed_commit::subxacts, xl_xact_parsed_commit::twophase_gid, xl_xact_parsed_commit::twophase_xid, UpdateDecodingStats(), xl_xact_parsed_commit::xact_time, and XLogRecGetOrigin.

Referenced by DecodeXactOp().

774 {
775  SnapBuild *builder = ctx->snapshot_builder;
776  XLogRecPtr origin_lsn = parsed->origin_lsn;
777  TimestampTz prepare_time = parsed->xact_time;
778  XLogRecPtr origin_id = XLogRecGetOrigin(buf->record);
779  int i;
780  TransactionId xid = parsed->twophase_xid;
781 
782  if (parsed->origin_timestamp != 0)
783  prepare_time = parsed->origin_timestamp;
784 
785  /*
786  * Remember the prepare info for a txn so that it can be used later in
787  * commit prepared if required. See ReorderBufferFinishPrepared.
788  */
789  if (!ReorderBufferRememberPrepareInfo(ctx->reorder, xid, buf->origptr,
790  buf->endptr, prepare_time, origin_id,
791  origin_lsn))
792  return;
793 
794  /* We can't start streaming unless a consistent state is reached. */
796  {
798  return;
799  }
800 
801  /*
802  * Check whether we need to process this transaction. See
803  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
804  * transaction.
805  *
806  * We can't call ReorderBufferForget as we did in DecodeCommit as the txn
807  * hasn't yet been committed, removing this txn before a commit might
808  * result in the computation of an incorrect restart_lsn. See
809  * SnapBuildProcessRunningXacts. But we need to process cache
810  * invalidations if there are any for the reasons mentioned in
811  * DecodeCommit.
812  */
813  if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
814  {
816  ReorderBufferInvalidate(ctx->reorder, xid, buf->origptr);
817  return;
818  }
819 
820  /* Tell the reorderbuffer about the surviving subtransactions. */
821  for (i = 0; i < parsed->nsubxacts; i++)
822  {
823  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
824  buf->origptr, buf->endptr);
825  }
826 
827  /* replay actions of all transaction + subtransactions in order */
828  ReorderBufferPrepare(ctx->reorder, xid, parsed->twophase_gid);
829 
830  /*
831  * Update the decoding stats at transaction prepare/commit/abort.
832  * Additionally we send the stats when we spill or stream the changes to
833  * avoid losing them in case the decoding is interrupted. It is not clear
834  * that sending more or less frequently than this would be better.
835  */
836  UpdateDecodingStats(ctx);
837 }
struct ReorderBuffer * reorder
Definition: logical.h:42
uint32 TransactionId
Definition: c.h:587
TimestampTz origin_timestamp
Definition: xact.h:346
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
int64 TimestampTz
Definition: timestamp.h:39
TransactionId twophase_xid
Definition: xact.h:340
TransactionId * subxacts
Definition: xact.h:332
XLogRecPtr origin_lsn
Definition: xact.h:345
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
XLogRecPtr endptr
Definition: decode.c:49
void ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid)
XLogReaderState * record
Definition: decode.c:50
static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid dbId, RepOriginId origin_id)
Definition: decode.c:1308
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
void UpdateDecodingStats(LogicalDecodingContext *ctx)
Definition: logical.c:1803
struct SnapBuild * snapshot_builder
Definition: logical.h:43
uint64 XLogRecPtr
Definition: xlogdefs.h:21
bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid, XLogRecPtr prepare_lsn, XLogRecPtr end_lsn, TimestampTz prepare_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
XLogRecPtr origptr
Definition: decode.c:48
int i
void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, char *gid)
void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
char twophase_gid[GIDSIZE]
Definition: xact.h:341
TimestampTz xact_time
Definition: xact.h:325

◆ DecodeSpecConfirm()

static void DecodeSpecConfirm ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

1227 {
1228  XLogReaderState *r = buf->record;
1229  ReorderBufferChange *change;
1230  RelFileNode target_node;
1231 
1232  /* only interested in our database */
1233  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
1234  if (target_node.dbNode != ctx->slot->data.database)
1235  return;
1236 
1237  /* output plugin doesn't look for this origin, no need to queue */
1238  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1239  return;
1240 
1241  change = ReorderBufferGetChange(ctx->reorder);
1243  change->origin_id = XLogRecGetOrigin(r);
1244 
1245  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1246 
1247  change->data.tp.clear_toast_afterwards = true;
1248 
1250  change, false);
1251 }
struct ReorderBuffer * reorder
Definition: logical.h:42
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:147
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:610
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:85
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
XLogReaderState * record
Definition: decode.c:50
union ReorderBufferChange::@97 data
RepOriginId origin_id
Definition: reorderbuffer.h:90
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1531
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
struct ReorderBufferChange::@97::@98 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 394 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().

395 {
396  SnapBuild *builder = ctx->snapshot_builder;
397  XLogReaderState *r = buf->record;
398  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
399 
401 
402  switch (info)
403  {
404  case XLOG_RUNNING_XACTS:
405  {
407 
408  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
409 
410  /*
411  * Abort all transactions that we keep track of, that are
412  * older than the record's oldestRunningXid. This is the most
413  * convenient spot for doing so since, in contrast to shutdown
414  * or end-of-recovery checkpoints, we have information about
415  * all running transactions which includes prepared ones,
416  * while shutdown checkpoints just know that no non-prepared
417  * transactions are in progress.
418  */
420  }
421  break;
422  case XLOG_STANDBY_LOCK:
423  break;
424  case XLOG_INVALIDATIONS:
425 
426  /*
427  * We are processing the invalidations at the command level via
428  * XLOG_XACT_INVALIDATIONS. So we don't need to do anything here.
429  */
430  break;
431  default:
432  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
433  }
434 }
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:439
#define XLOG_STANDBY_LOCK
Definition: standbydefs.h:34
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
#define ERROR
Definition: elog.h:46
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1104
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:315
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
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:232
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 1082 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().

1083 {
1084  XLogReaderState *r = buf->record;
1085  xl_heap_truncate *xlrec;
1086  ReorderBufferChange *change;
1087 
1088  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
1089 
1090  /* only interested in our database */
1091  if (xlrec->dbId != ctx->slot->data.database)
1092  return;
1093 
1094  /* output plugin doesn't look for this origin, no need to queue */
1095  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1096  return;
1097 
1098  change = ReorderBufferGetChange(ctx->reorder);
1100  change->origin_id = XLogRecGetOrigin(r);
1101  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
1102  change->data.truncate.cascade = true;
1103  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
1104  change->data.truncate.restart_seqs = true;
1105  change->data.truncate.nrelids = xlrec->nrelids;
1106  change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
1107  xlrec->nrelids);
1108  memcpy(change->data.truncate.relids, xlrec->relids,
1109  xlrec->nrelids * sizeof(Oid));
1111  buf->origptr, change, false);
1112 }
#define XLH_TRUNCATE_CASCADE
Definition: heapam_xlog.h:120
struct ReorderBuffer * reorder
Definition: logical.h:42
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:147
unsigned int Oid
Definition: postgres_ext.h:31
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:610
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:85
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
Oid relids[FLEXIBLE_ARRAY_MEMBER]
Definition: heapam_xlog.h:133
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
XLogReaderState * record
Definition: decode.c:50
union ReorderBufferChange::@97 data
RepOriginId origin_id
Definition: reorderbuffer.h:90
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
#define XLH_TRUNCATE_RESTART_SEQS
Definition: heapam_xlog.h:121
XLogRecPtr origptr
Definition: decode.c:48
struct ReorderBufferChange::@97::@99 truncate
ReplicationSlot * slot
Definition: logical.h:38
Oid * ReorderBufferGetRelids(ReorderBuffer *rb, int nrelids)

◆ DecodeTXNNeedSkip()

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

Definition at line 1308 of file decode.c.

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

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

1310 {
1311  return (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
1312  (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) ||
1313  ctx->fast_forward || FilterByOrigin(ctx, origin_id));
1314 }
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:394
ReplicationSlotPersistentData data
Definition: slot.h:147
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:610
#define InvalidOid
Definition: postgres_ext.h:36
struct SnapBuild * snapshot_builder
Definition: logical.h:43
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38

◆ DecodeUpdate()

static void DecodeUpdate ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

962 {
963  XLogReaderState *r = buf->record;
964  xl_heap_update *xlrec;
965  ReorderBufferChange *change;
966  char *data;
967  RelFileNode target_node;
968 
969  xlrec = (xl_heap_update *) XLogRecGetData(r);
970 
971  /* only interested in our database */
972  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
973  if (target_node.dbNode != ctx->slot->data.database)
974  return;
975 
976  /* output plugin doesn't look for this origin, no need to queue */
977  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
978  return;
979 
980  change = ReorderBufferGetChange(ctx->reorder);
982  change->origin_id = XLogRecGetOrigin(r);
983  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
984 
986  {
987  Size datalen;
988  Size tuplelen;
989 
990  data = XLogRecGetBlockData(r, 0, &datalen);
991 
992  tuplelen = datalen - SizeOfHeapHeader;
993 
994  change->data.tp.newtuple =
995  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
996 
997  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
998  }
999 
1000  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
1001  {
1002  Size datalen;
1003  Size tuplelen;
1004 
1005  /* caution, remaining data in record is not aligned */
1006  data = XLogRecGetData(r) + SizeOfHeapUpdate;
1007  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
1008  tuplelen = datalen - SizeOfHeapHeader;
1009 
1010  change->data.tp.oldtuple =
1011  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1012 
1013  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
1014  }
1015 
1016  change->data.tp.clear_toast_afterwards = true;
1017 
1019  change, false);
1020 }
struct ReorderBuffer * reorder
Definition: logical.h:42
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1262
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:147
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:610
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:85
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:318
#define XLH_UPDATE_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:84
#define SizeOfHeapUpdate
Definition: heapam_xlog.h:227
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:321
XLogReaderState * record
Definition: decode.c:50
union ReorderBufferChange::@97 data
RepOriginId origin_id
Definition: reorderbuffer.h:90
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1531
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1555
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
size_t Size
Definition: c.h:540
struct ReorderBufferChange::@97::@98 tp
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define XLH_UPDATE_CONTAINS_OLD
Definition: heapam_xlog.h:89
#define SizeOfHeapHeader
Definition: heapam_xlog.h:151
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeXactOp()

static void DecodeXactOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 237 of file decode.c.

References DecodeAbort(), DecodeCommit(), DecodePrepare(), elog, ERROR, LogicalDecodingContext::fast_forward, FilterPrepare(), xl_xact_invals::msgs, xl_xact_invals::nmsgs, XLogRecordBuffer::origptr, ParseAbortRecord(), ParseCommitRecord(), ParsePrepareRecord(), XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAddInvalidations(), ReorderBufferImmediateInvalidation(), ReorderBufferProcessXid(), ReorderBufferXidSetCatalogChanges(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), LogicalDecodingContext::snapshot_builder, TransactionIdIsValid, two_phase, xl_xact_parsed_commit::twophase_gid, xl_xact_parsed_abort::twophase_gid, xl_xact_parsed_commit::twophase_xid, xl_xact_parsed_abort::twophase_xid, XLOG_XACT_ABORT, XLOG_XACT_ABORT_PREPARED, XLOG_XACT_ASSIGNMENT, XLOG_XACT_COMMIT, XLOG_XACT_COMMIT_PREPARED, XLOG_XACT_INVALIDATIONS, XLOG_XACT_OPMASK, XLOG_XACT_PREPARE, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

Referenced by LogicalDecodingProcessRecord().

238 {
239  SnapBuild *builder = ctx->snapshot_builder;
240  ReorderBuffer *reorder = ctx->reorder;
241  XLogReaderState *r = buf->record;
243 
244  /*
245  * If the snapshot isn't yet fully built, we cannot decode anything, so
246  * bail out.
247  */
249  return;
250 
251  switch (info)
252  {
253  case XLOG_XACT_COMMIT:
255  {
256  xl_xact_commit *xlrec;
257  xl_xact_parsed_commit parsed;
258  TransactionId xid;
259  bool two_phase = false;
260 
261  xlrec = (xl_xact_commit *) XLogRecGetData(r);
262  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
263 
264  if (!TransactionIdIsValid(parsed.twophase_xid))
265  xid = XLogRecGetXid(r);
266  else
267  xid = parsed.twophase_xid;
268 
269  /*
270  * We would like to process the transaction in a two-phase
271  * manner iff output plugin supports two-phase commits and
272  * doesn't filter the transaction at prepare time.
273  */
274  if (info == XLOG_XACT_COMMIT_PREPARED)
275  two_phase = !(FilterPrepare(ctx, xid,
276  parsed.twophase_gid));
277 
278  DecodeCommit(ctx, buf, &parsed, xid, two_phase);
279  break;
280  }
281  case XLOG_XACT_ABORT:
283  {
284  xl_xact_abort *xlrec;
285  xl_xact_parsed_abort parsed;
286  TransactionId xid;
287  bool two_phase = false;
288 
289  xlrec = (xl_xact_abort *) XLogRecGetData(r);
290  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
291 
292  if (!TransactionIdIsValid(parsed.twophase_xid))
293  xid = XLogRecGetXid(r);
294  else
295  xid = parsed.twophase_xid;
296 
297  /*
298  * We would like to process the transaction in a two-phase
299  * manner iff output plugin supports two-phase commits and
300  * doesn't filter the transaction at prepare time.
301  */
302  if (info == XLOG_XACT_ABORT_PREPARED)
303  two_phase = !(FilterPrepare(ctx, xid,
304  parsed.twophase_gid));
305 
306  DecodeAbort(ctx, buf, &parsed, xid, two_phase);
307  break;
308  }
310 
311  /*
312  * We assign subxact to the toplevel xact while processing each
313  * record if required. So, we don't need to do anything here. See
314  * LogicalDecodingProcessRecord.
315  */
316  break;
318  {
319  TransactionId xid;
320  xl_xact_invals *invals;
321 
322  xid = XLogRecGetXid(r);
323  invals = (xl_xact_invals *) XLogRecGetData(r);
324 
325  /*
326  * Execute the invalidations for xid-less transactions,
327  * otherwise, accumulate them so that they can be processed at
328  * the commit time.
329  */
330  if (TransactionIdIsValid(xid))
331  {
332  if (!ctx->fast_forward)
333  ReorderBufferAddInvalidations(reorder, xid,
334  buf->origptr,
335  invals->nmsgs,
336  invals->msgs);
338  buf->origptr);
339  }
340  else if ((!ctx->fast_forward))
342  invals->nmsgs,
343  invals->msgs);
344  }
345  break;
346  case XLOG_XACT_PREPARE:
347  {
348  xl_xact_parsed_prepare parsed;
349  xl_xact_prepare *xlrec;
350 
351  /* ok, parse it */
352  xlrec = (xl_xact_prepare *) XLogRecGetData(r);
354  xlrec, &parsed);
355 
356  /*
357  * We would like to process the transaction in a two-phase
358  * manner iff output plugin supports two-phase commits and
359  * doesn't filter the transaction at prepare time.
360  */
361  if (FilterPrepare(ctx, parsed.twophase_xid,
362  parsed.twophase_gid))
363  {
364  ReorderBufferProcessXid(reorder, parsed.twophase_xid,
365  buf->origptr);
366  break;
367  }
368 
369  /*
370  * Note that if the prepared transaction has locked [user]
371  * catalog tables exclusively then decoding prepare can block
372  * till the main transaction is committed because it needs to
373  * lock the catalog tables.
374  *
375  * XXX Now, this can even lead to a deadlock if the prepare
376  * transaction is waiting to get it logically replicated for
377  * distributed 2PC. This can be avoided by disallowing
378  * preparing transactions that have locked [user] catalog
379  * tables exclusively but as of now, we ask users not to do
380  * such an operation.
381  */
382  DecodePrepare(ctx, buf, &parsed);
383  break;
384  }
385  default:
386  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
387  }
388 }
#define XLOG_XACT_COMMIT
Definition: xact.h:148
struct ReorderBuffer * reorder
Definition: logical.h:42
#define XLOG_XACT_PREPARE
Definition: xact.h:149
uint32 TransactionId
Definition: c.h:587
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
#define XLOG_XACT_INVALIDATIONS
Definition: xact.h:154
TransactionId twophase_xid
Definition: xact.h:340
unsigned char uint8
Definition: c.h:439
#define XLogRecGetData(decoder)
Definition: xlogreader.h:320
#define ERROR
Definition: elog.h:46
SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER]
Definition: xact.h:255
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:677
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:152
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:315
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:153
static bool two_phase
void ParsePrepareRecord(uint8 info, xl_xact_prepare *xlrec, xl_xact_parsed_prepare *parsed)
Definition: xactdesc.c:216
TransactionId twophase_xid
Definition: xact.h:365
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLOG_XACT_ABORT
Definition: xact.h:150
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
static bool FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: decode.c:587
#define XLOG_XACT_OPMASK
Definition: xact.h:158
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:232
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:151
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
Definition: decode.c:772
int nmsgs
Definition: xact.h:254
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:847
char twophase_gid[GIDSIZE]
Definition: xact.h:341
char twophase_gid[GIDSIZE]
Definition: xact.h:366

◆ DecodeXLogOp()

static void DecodeXLogOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

196 {
197  SnapBuild *builder = ctx->snapshot_builder;
198  uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
199 
201  buf->origptr);
202 
203  switch (info)
204  {
205  /* this is also used in END_OF_RECOVERY checkpoints */
208  SnapBuildSerializationPoint(builder, buf->origptr);
209 
210  break;
212 
213  /*
214  * a RUNNING_XACTS record will have been logged near to this, we
215  * can restart from there.
216  */
217  break;
218  case XLOG_NOOP:
219  case XLOG_NEXTOID:
220  case XLOG_SWITCH:
221  case XLOG_BACKUP_END:
223  case XLOG_RESTORE_POINT:
224  case XLOG_FPW_CHANGE:
225  case XLOG_FPI_FOR_HINT:
226  case XLOG_FPI:
227  break;
228  default:
229  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
230  }
231 }
#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:439
#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:46
#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:315
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
#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:232
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1480
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)

◆ DecodeXLogTuple()

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

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

1263 {
1264  xl_heap_header xlhdr;
1265  int datalen = len - SizeOfHeapHeader;
1267 
1268  Assert(datalen >= 0);
1269 
1270  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1271  header = tuple->tuple.t_data;
1272 
1273  /* not a disk based tuple */
1274  ItemPointerSetInvalid(&tuple->tuple.t_self);
1275 
1276  /* we can only figure this out after reassembling the transactions */
1277  tuple->tuple.t_tableOid = InvalidOid;
1278 
1279  /* data is not stored aligned, copy to aligned storage */
1280  memcpy((char *) &xlhdr,
1281  data,
1282  SizeOfHeapHeader);
1283 
1284  memset(header, 0, SizeofHeapTupleHeader);
1285 
1286  memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
1287  data + SizeOfHeapHeader,
1288  datalen);
1289 
1290  header->t_infomask = xlhdr.t_infomask;
1291  header->t_infomask2 = xlhdr.t_infomask2;
1292  header->t_hoff = xlhdr.t_hoff;
1293 }
#define SizeofHeapTupleHeader
Definition: htup_details.h:184
uint16 t_infomask2
Definition: heapam_xlog.h:146
#define InvalidOid
Definition: postgres_ext.h:36
#define Assert(condition)
Definition: c.h:804
uint16 t_infomask
Definition: heapam_xlog.h:147
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:212
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:172
#define SizeOfHeapHeader
Definition: heapam_xlog.h:151

◆ FilterByOrigin()

static bool FilterByOrigin ( LogicalDecodingContext ctx,
RepOriginId  origin_id 
)
inlinestatic

Definition at line 610 of file decode.c.

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

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

611 {
612  if (ctx->callbacks.filter_by_origin_cb == NULL)
613  return false;
614 
615  return filter_by_origin_cb_wrapper(ctx, origin_id);
616 }
OutputPluginCallbacks callbacks
Definition: logical.h:52
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:1140
LogicalDecodeFilterByOriginCB filter_by_origin_cb

◆ FilterPrepare()

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

Definition at line 587 of file decode.c.

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

Referenced by DecodeXactOp().

589 {
590  /*
591  * Skip if decoding of two-phase transactions at PREPARE time is not
592  * enabled. In that case, all two-phase transactions are considered
593  * filtered out and will be applied as regular transactions at COMMIT
594  * PREPARED.
595  */
596  if (!ctx->twophase)
597  return true;
598 
599  /*
600  * The filter_prepare callback is optional. When not supplied, all
601  * prepared transactions should go through.
602  */
603  if (ctx->callbacks.filter_prepare_cb == NULL)
604  return false;
605 
606  return filter_prepare_cb_wrapper(ctx, xid, gid);
607 }
bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: logical.c:1109
OutputPluginCallbacks callbacks
Definition: logical.h:52
LogicalDecodeFilterPrepareCB filter_prepare_cb

◆ LogicalDecodingProcessRecord()

void LogicalDecodingProcessRecord ( LogicalDecodingContext ctx,
XLogReaderState record 
)

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

107 {
109  TransactionId txid;
110 
111  buf.origptr = ctx->reader->ReadRecPtr;
112  buf.endptr = ctx->reader->EndRecPtr;
113  buf.record = record;
114 
115  txid = XLogRecGetTopXid(record);
116 
117  /*
118  * If the top-level xid is valid, we need to assign the subxact to the
119  * top-level xact. We need to do this for all records, hence we do it
120  * before the switch.
121  */
122  if (TransactionIdIsValid(txid))
123  {
125  txid,
126  record->decoded_record->xl_xid,
127  buf.origptr);
128  }
129 
130  /* cast so we get a warning when new rmgrs are added */
131  switch ((RmgrId) XLogRecGetRmid(record))
132  {
133  /*
134  * Rmgrs we care about for logical decoding. Add new rmgrs in
135  * rmgrlist.h's order.
136  */
137  case RM_XLOG_ID:
138  DecodeXLogOp(ctx, &buf);
139  break;
140 
141  case RM_XACT_ID:
142  DecodeXactOp(ctx, &buf);
143  break;
144 
145  case RM_STANDBY_ID:
146  DecodeStandbyOp(ctx, &buf);
147  break;
148 
149  case RM_HEAP2_ID:
150  DecodeHeap2Op(ctx, &buf);
151  break;
152 
153  case RM_HEAP_ID:
154  DecodeHeapOp(ctx, &buf);
155  break;
156 
157  case RM_LOGICALMSG_ID:
158  DecodeLogicalMsgOp(ctx, &buf);
159  break;
160 
161  /*
162  * Rmgrs irrelevant for logical decoding; they describe stuff not
163  * represented in logical decoding. Add new rmgrs in rmgrlist.h's
164  * order.
165  */
166  case RM_SMGR_ID:
167  case RM_CLOG_ID:
168  case RM_DBASE_ID:
169  case RM_TBLSPC_ID:
170  case RM_MULTIXACT_ID:
171  case RM_RELMAP_ID:
172  case RM_BTREE_ID:
173  case RM_HASH_ID:
174  case RM_GIN_ID:
175  case RM_GIST_ID:
176  case RM_SEQ_ID:
177  case RM_SPGIST_ID:
178  case RM_BRIN_ID:
179  case RM_COMMIT_TS_ID:
180  case RM_REPLORIGIN_ID:
181  case RM_GENERIC_ID:
182  /* just deal with xid, and done */
184  buf.origptr);
185  break;
186  case RM_NEXT_ID:
187  elog(ERROR, "unexpected RM_NEXT_ID rmgr_id: %u", (RmgrIds) XLogRecGetRmid(buf.record));
188  }
189 }
struct ReorderBuffer * reorder
Definition: logical.h:42
uint32 TransactionId
Definition: c.h:587
XLogRecPtr EndRecPtr
Definition: xlogreader.h:176
static void DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:237
static void DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:622
#define ERROR
Definition: elog.h:46
XLogRecPtr endptr
Definition: decode.c:49
static char * buf
Definition: pg_test_fsync.c:68
XLogRecPtr ReadRecPtr
Definition: xlogreader.h:175
XLogRecord * decoded_record
Definition: xlogreader.h:196
XLogReaderState * record
Definition: decode.c:50
static void DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:195
void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
static void DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:440
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:317
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:232
static void DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:500
#define XLogRecGetTopXid(decoder)
Definition: xlogreader.h:319
#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:394
#define XLogRecGetRmid(decoder)
Definition: xlogreader.h:316