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

Go to the source code of this file.

Functions

static void DecodeInsert (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeUpdate (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeDelete (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeTruncate (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeMultiInsert (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeSpecConfirm (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeCommit (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid, bool two_phase)
 
static void DecodeAbort (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
 
static void DecodePrepare (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
 
static void DecodeXLogTuple (char *data, Size len, 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)
 
void xlog_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void xact_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void standby_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void heap2_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void heap_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static bool FilterByOrigin (LogicalDecodingContext *ctx, RepOriginId origin_id)
 
void logicalmsg_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 

Function Documentation

◆ DecodeAbort()

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

Definition at line 785 of file decode.c.

788 {
789  int i;
790  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
791  TimestampTz abort_time = parsed->xact_time;
792  XLogRecPtr origin_id = XLogRecGetOrigin(buf->record);
793  bool skip_xact;
794 
795  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
796  {
797  origin_lsn = parsed->origin_lsn;
798  abort_time = parsed->origin_timestamp;
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  skip_xact = DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id);
807 
808  /*
809  * Send the final rollback record for a prepared transaction unless we
810  * need to skip it. For non-two-phase xacts, simply forget the xact.
811  */
812  if (two_phase && !skip_xact)
813  {
814  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
816  abort_time, origin_id, origin_lsn,
817  parsed->twophase_gid, false);
818  }
819  else
820  {
821  for (i = 0; i < parsed->nsubxacts; i++)
822  {
823  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
824  buf->record->EndRecPtr);
825  }
826 
827  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
828  }
829 
830  /* update the decoding stats */
831  UpdateDecodingStats(ctx);
832 }
int64 TimestampTz
Definition: timestamp.h:39
static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid dbId, RepOriginId origin_id)
Definition: decode.c:1246
int i
Definition: isn.c:73
void UpdateDecodingStats(LogicalDecodingContext *ctx)
Definition: logical.c:1833
static bool two_phase
static char * buf
Definition: pg_test_fsync.c:67
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)
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
struct ReorderBuffer * reorder
Definition: logical.h:43
TimestampTz xact_time
Definition: xact.h:393
TransactionId * subxacts
Definition: xact.h:400
XLogRecPtr origin_lsn
Definition: xact.h:411
char twophase_gid[GIDSIZE]
Definition: xact.h:409
TimestampTz origin_timestamp
Definition: xact.h:412
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:180
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:411

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

Referenced by xact_decode().

◆ DecodeCommit()

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

Definition at line 615 of file decode.c.

618 {
619  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
620  TimestampTz commit_time = parsed->xact_time;
621  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
622  int i;
623 
624  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
625  {
626  origin_lsn = parsed->origin_lsn;
627  commit_time = parsed->origin_timestamp;
628  }
629 
630  SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
631  parsed->nsubxacts, parsed->subxacts);
632 
633  /* ----
634  * Check whether we are interested in this specific transaction, and tell
635  * the reorderbuffer to forget the content of the (sub-)transactions
636  * if not.
637  *
638  * We can't just use ReorderBufferAbort() here, because we need to execute
639  * the transaction's invalidations. This currently won't be needed if
640  * we're just skipping over the transaction because currently we only do
641  * so during startup, to get to the first transaction the client needs. As
642  * we have reset the catalog caches before starting to read WAL, and we
643  * haven't yet touched any catalogs, there can't be anything to invalidate.
644  * But if we're "forgetting" this commit because it happened in another
645  * database, the invalidations might be important, because they could be
646  * for shared catalogs and we might have loaded data into the relevant
647  * syscaches.
648  * ---
649  */
650  if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
651  {
652  for (i = 0; i < parsed->nsubxacts; i++)
653  {
654  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
655  }
656  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
657 
658  return;
659  }
660 
661  /* tell the reorderbuffer about the surviving subtransactions */
662  for (i = 0; i < parsed->nsubxacts; i++)
663  {
664  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
665  buf->origptr, buf->endptr);
666  }
667 
668  /*
669  * Send the final commit record if the transaction data is already
670  * decoded, otherwise, process the entire transaction.
671  */
672  if (two_phase)
673  {
674  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
676  commit_time, origin_id, origin_lsn,
677  parsed->twophase_gid, true);
678  }
679  else
680  {
681  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
682  commit_time, origin_id, origin_lsn);
683  }
684 
685  /*
686  * Update the decoding stats at transaction prepare/commit/abort.
687  * Additionally we send the stats when we spill or stream the changes to
688  * avoid losing them in case the decoding is interrupted. It is not clear
689  * that sending more or less frequently than this would be better.
690  */
691  UpdateDecodingStats(ctx);
692 }
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
XLogRecPtr SnapBuildGetTwoPhaseAt(SnapBuild *builder)
Definition: snapbuild.c:376
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:937
struct SnapBuild * snapshot_builder
Definition: logical.h:44
TimestampTz xact_time
Definition: xact.h:360
TimestampTz origin_timestamp
Definition: xact.h:386
TransactionId * subxacts
Definition: xact.h:367
char twophase_gid[GIDSIZE]
Definition: xact.h:379
XLogRecPtr origin_lsn
Definition: xact.h:385
uint16 RepOriginId
Definition: xlogdefs.h:65

References buf, xl_xact_parsed_commit::dbId, DecodeTXNNeedSkip(), i, InvalidXLogRecPtr, xl_xact_parsed_commit::nsubxacts, xl_xact_parsed_commit::origin_lsn, xl_xact_parsed_commit::origin_timestamp, LogicalDecodingContext::reorder, ReorderBufferCommit(), ReorderBufferCommitChild(), ReorderBufferFinishPrepared(), ReorderBufferForget(), SnapBuildCommitTxn(), SnapBuildGetTwoPhaseAt(), LogicalDecodingContext::snapshot_builder, xl_xact_parsed_commit::subxacts, two_phase, 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 xact_decode().

◆ DecodeDelete()

static void DecodeDelete ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 966 of file decode.c.

967 {
968  XLogReaderState *r = buf->record;
969  xl_heap_delete *xlrec;
970  ReorderBufferChange *change;
971  RelFileNode target_node;
972 
973  xlrec = (xl_heap_delete *) XLogRecGetData(r);
974 
975  /* only interested in our database */
976  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
977  if (target_node.dbNode != ctx->slot->data.database)
978  return;
979 
980  /* output plugin doesn't look for this origin, no need to queue */
981  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
982  return;
983 
984  change = ReorderBufferGetChange(ctx->reorder);
985 
986  if (xlrec->flags & XLH_DELETE_IS_SUPER)
988  else
990 
991  change->origin_id = XLogRecGetOrigin(r);
992 
993  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
994 
995  /* old primary key stored */
996  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
997  {
998  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
999  Size tuplelen = datalen - SizeOfHeapHeader;
1000 
1002 
1003  change->data.tp.oldtuple =
1004  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1005 
1006  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
1007  datalen, change->data.tp.oldtuple);
1008  }
1009 
1010  change->data.tp.clear_toast_afterwards = true;
1011 
1013  change, false);
1014 }
size_t Size
Definition: c.h:540
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1200
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:548
#define XLH_DELETE_CONTAINS_OLD
Definition: heapam_xlog.h:103
#define SizeOfHeapHeader
Definition: heapam_xlog.h:151
#define SizeOfHeapDelete
Definition: heapam_xlog.h:115
#define XLH_DELETE_IS_SUPER
Definition: heapam_xlog.h:99
Assert(fmt[strlen(fmt) - 1] !='\n')
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT
Definition: reorderbuffer.h:66
@ REORDER_BUFFER_CHANGE_DELETE
Definition: reorderbuffer.h:58
ReplicationSlot * slot
Definition: logical.h:39
ReorderBufferChangeType action
Definition: reorderbuffer.h:85
struct ReorderBufferChange::@105::@106 tp
union ReorderBufferChange::@105 data
RepOriginId origin_id
Definition: reorderbuffer.h:90
ReplicationSlotPersistentData data
Definition: slot.h:147
void XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1928
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:414
#define XLogRecGetData(decoder)
Definition: xlogreader.h:413
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:410

References ReorderBufferChange::action, Assert(), buf, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_delete::flags, ReorderBufferChange::origin_id, 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 heap_decode().

◆ DecodeInsert()

static void DecodeInsert ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 840 of file decode.c.

841 {
842  Size datalen;
843  char *tupledata;
844  Size tuplelen;
845  XLogReaderState *r = buf->record;
846  xl_heap_insert *xlrec;
847  ReorderBufferChange *change;
848  RelFileNode target_node;
849 
850  xlrec = (xl_heap_insert *) XLogRecGetData(r);
851 
852  /*
853  * Ignore insert records without new tuples (this does happen when
854  * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
855  */
856  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
857  return;
858 
859  /* only interested in our database */
860  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
861  if (target_node.dbNode != ctx->slot->data.database)
862  return;
863 
864  /* output plugin doesn't look for this origin, no need to queue */
865  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
866  return;
867 
868  change = ReorderBufferGetChange(ctx->reorder);
869  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
871  else
873  change->origin_id = XLogRecGetOrigin(r);
874 
875  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
876 
877  tupledata = XLogRecGetBlockData(r, 0, &datalen);
878  tuplelen = datalen - SizeOfHeapHeader;
879 
880  change->data.tp.newtuple =
881  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
882 
883  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
884 
885  change->data.tp.clear_toast_afterwards = true;
886 
888  change,
890 }
#define XLH_INSERT_ON_TOAST_RELATION
Definition: heapam_xlog.h:70
#define XLH_INSERT_IS_SPECULATIVE
Definition: heapam_xlog.h:68
#define XLH_INSERT_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:69
@ REORDER_BUFFER_CHANGE_INSERT
Definition: reorderbuffer.h:56
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT
Definition: reorderbuffer.h:64
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1981

References ReorderBufferChange::action, buf, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_insert::flags, ReorderBufferChange::origin_id, 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 heap_decode().

◆ DecodeMultiInsert()

static void DecodeMultiInsert ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 1058 of file decode.c.

1059 {
1060  XLogReaderState *r = buf->record;
1061  xl_heap_multi_insert *xlrec;
1062  int i;
1063  char *data;
1064  char *tupledata;
1065  Size tuplelen;
1066  RelFileNode rnode;
1067 
1068  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
1069 
1070  /*
1071  * Ignore insert records without new tuples. This happens when a
1072  * multi_insert is done on a catalog or on a non-persistent relation.
1073  */
1074  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
1075  return;
1076 
1077  /* only interested in our database */
1078  XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
1079  if (rnode.dbNode != ctx->slot->data.database)
1080  return;
1081 
1082  /* output plugin doesn't look for this origin, no need to queue */
1083  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1084  return;
1085 
1086  /*
1087  * We know that this multi_insert isn't for a catalog, so the block should
1088  * always have data even if a full-page write of it is taken.
1089  */
1090  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
1091  Assert(tupledata != NULL);
1092 
1093  data = tupledata;
1094  for (i = 0; i < xlrec->ntuples; i++)
1095  {
1096  ReorderBufferChange *change;
1097  xl_multi_insert_tuple *xlhdr;
1098  int datalen;
1099  ReorderBufferTupleBuf *tuple;
1101 
1102  change = ReorderBufferGetChange(ctx->reorder);
1104  change->origin_id = XLogRecGetOrigin(r);
1105 
1106  memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
1107 
1108  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
1109  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
1110  datalen = xlhdr->datalen;
1111 
1112  change->data.tp.newtuple =
1113  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
1114 
1115  tuple = change->data.tp.newtuple;
1116  header = tuple->tuple.t_data;
1117 
1118  /* not a disk based tuple */
1120 
1121  /*
1122  * We can only figure this out after reassembling the transactions.
1123  */
1124  tuple->tuple.t_tableOid = InvalidOid;
1125 
1126  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1127 
1128  memset(header, 0, SizeofHeapTupleHeader);
1129 
1130  memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
1131  (char *) data,
1132  datalen);
1133  header->t_infomask = xlhdr->t_infomask;
1134  header->t_infomask2 = xlhdr->t_infomask2;
1135  header->t_hoff = xlhdr->t_hoff;
1136 
1137  /*
1138  * Reset toast reassembly state only after the last row in the last
1139  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
1140  * call.
1141  */
1142  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
1143  (i + 1) == xlrec->ntuples)
1144  change->data.tp.clear_toast_afterwards = true;
1145  else
1146  change->data.tp.clear_toast_afterwards = false;
1147 
1149  buf->origptr, change, false);
1150 
1151  /* move to the next xl_multi_insert_tuple entry */
1152  data += datalen;
1153  }
1154  Assert(data == tupledata + tuplelen);
1155 }
#define SHORTALIGN(LEN)
Definition: c.h:753
#define XLH_INSERT_LAST_IN_MULTI
Definition: heapam_xlog.h:67
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:193
#define SizeofHeapTupleHeader
Definition: htup_details.h:184
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:172
const void * data
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:212
#define InvalidOid
Definition: postgres_ext.h:36
ItemPointerData t_self
Definition: htup.h:65
uint32 t_len
Definition: htup.h:64
HeapTupleHeader t_data
Definition: htup.h:68
Oid t_tableOid
Definition: htup.h:66
HeapTupleData tuple
Definition: reorderbuffer.h:29

References ReorderBufferChange::action, Assert(), buf, ReorderBufferChange::data, ReplicationSlot::data, 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, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INSERT, ReorderBufferGetChange(), ReorderBufferGetTupleBuf(), ReorderBufferQueueChange(), SHORTALIGN, SizeofHeapTupleHeader, SizeOfMultiInsertTuple, LogicalDecodingContext::slot, HeapTupleData::t_data, xl_multi_insert_tuple::t_hoff, xl_multi_insert_tuple::t_infomask, 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 heap2_decode().

◆ DecodePrepare()

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

Definition at line 710 of file decode.c.

712 {
713  SnapBuild *builder = ctx->snapshot_builder;
714  XLogRecPtr origin_lsn = parsed->origin_lsn;
715  TimestampTz prepare_time = parsed->xact_time;
716  XLogRecPtr origin_id = XLogRecGetOrigin(buf->record);
717  int i;
718  TransactionId xid = parsed->twophase_xid;
719 
720  if (parsed->origin_timestamp != 0)
721  prepare_time = parsed->origin_timestamp;
722 
723  /*
724  * Remember the prepare info for a txn so that it can be used later in
725  * commit prepared if required. See ReorderBufferFinishPrepared.
726  */
727  if (!ReorderBufferRememberPrepareInfo(ctx->reorder, xid, buf->origptr,
728  buf->endptr, prepare_time, origin_id,
729  origin_lsn))
730  return;
731 
732  /* We can't start streaming unless a consistent state is reached. */
734  {
736  return;
737  }
738 
739  /*
740  * Check whether we need to process this transaction. See
741  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
742  * transaction.
743  *
744  * We can't call ReorderBufferForget as we did in DecodeCommit as the txn
745  * hasn't yet been committed, removing this txn before a commit might
746  * result in the computation of an incorrect restart_lsn. See
747  * SnapBuildProcessRunningXacts. But we need to process cache
748  * invalidations if there are any for the reasons mentioned in
749  * DecodeCommit.
750  */
751  if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
752  {
754  ReorderBufferInvalidate(ctx->reorder, xid, buf->origptr);
755  return;
756  }
757 
758  /* Tell the reorderbuffer about the surviving subtransactions. */
759  for (i = 0; i < parsed->nsubxacts; i++)
760  {
761  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
762  buf->origptr, buf->endptr);
763  }
764 
765  /* replay actions of all transaction + subtransactions in order */
766  ReorderBufferPrepare(ctx->reorder, xid, parsed->twophase_gid);
767 
768  /*
769  * Update the decoding stats at transaction prepare/commit/abort.
770  * Additionally we send the stats when we spill or stream the changes to
771  * avoid losing them in case the decoding is interrupted. It is not clear
772  * that sending more or less frequently than this would be better.
773  */
774  UpdateDecodingStats(ctx);
775 }
uint32 TransactionId
Definition: c.h:587
void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, char *gid)
void ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid)
bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid, XLogRecPtr prepare_lsn, XLogRecPtr end_lsn, TimestampTz prepare_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:46
TransactionId twophase_xid
Definition: xact.h:378

References buf, xl_xact_parsed_commit::dbId, DecodeTXNNeedSkip(), i, xl_xact_parsed_commit::nsubxacts, xl_xact_parsed_commit::origin_lsn, xl_xact_parsed_commit::origin_timestamp, 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 xact_decode().

◆ DecodeSpecConfirm()

static void DecodeSpecConfirm ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 1164 of file decode.c.

1165 {
1166  XLogReaderState *r = buf->record;
1167  ReorderBufferChange *change;
1168  RelFileNode target_node;
1169 
1170  /* only interested in our database */
1171  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
1172  if (target_node.dbNode != ctx->slot->data.database)
1173  return;
1174 
1175  /* output plugin doesn't look for this origin, no need to queue */
1176  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1177  return;
1178 
1179  change = ReorderBufferGetChange(ctx->reorder);
1181  change->origin_id = XLogRecGetOrigin(r);
1182 
1183  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1184 
1185  change->data.tp.clear_toast_afterwards = true;
1186 
1188  change, false);
1189 }
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM
Definition: reorderbuffer.h:65

References ReorderBufferChange::action, buf, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, FilterByOrigin(), ReorderBufferChange::origin_id, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, ReorderBufferGetChange(), ReorderBufferQueueChange(), LogicalDecodingContext::slot, ReorderBufferChange::tp, XLogRecGetBlockTag(), XLogRecGetOrigin, and XLogRecGetXid.

Referenced by heap_decode().

◆ DecodeTruncate()

static void DecodeTruncate ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 1020 of file decode.c.

1021 {
1022  XLogReaderState *r = buf->record;
1023  xl_heap_truncate *xlrec;
1024  ReorderBufferChange *change;
1025 
1026  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
1027 
1028  /* only interested in our database */
1029  if (xlrec->dbId != ctx->slot->data.database)
1030  return;
1031 
1032  /* output plugin doesn't look for this origin, no need to queue */
1033  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1034  return;
1035 
1036  change = ReorderBufferGetChange(ctx->reorder);
1038  change->origin_id = XLogRecGetOrigin(r);
1039  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
1040  change->data.truncate.cascade = true;
1041  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
1042  change->data.truncate.restart_seqs = true;
1043  change->data.truncate.nrelids = xlrec->nrelids;
1044  change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
1045  xlrec->nrelids);
1046  memcpy(change->data.truncate.relids, xlrec->relids,
1047  xlrec->nrelids * sizeof(Oid));
1049  buf->origptr, change, false);
1050 }
#define XLH_TRUNCATE_RESTART_SEQS
Definition: heapam_xlog.h:121
#define XLH_TRUNCATE_CASCADE
Definition: heapam_xlog.h:120
unsigned int Oid
Definition: postgres_ext.h:31
Oid * ReorderBufferGetRelids(ReorderBuffer *rb, int nrelids)
@ REORDER_BUFFER_CHANGE_TRUNCATE
Definition: reorderbuffer.h:67
struct ReorderBufferChange::@105::@107 truncate
Oid relids[FLEXIBLE_ARRAY_MEMBER]
Definition: heapam_xlog.h:133

References ReorderBufferChange::action, buf, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, xl_heap_truncate::dbId, FilterByOrigin(), xl_heap_truncate::flags, xl_heap_truncate::nrelids, ReorderBufferChange::origin_id, 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 heap_decode().

◆ DecodeTXNNeedSkip()

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

Definition at line 1246 of file decode.c.

1248 {
1249  return (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
1250  (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) ||
1251  ctx->fast_forward || FilterByOrigin(ctx, origin_id));
1252 }
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:394

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

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

◆ DecodeUpdate()

static void DecodeUpdate ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 899 of file decode.c.

900 {
901  XLogReaderState *r = buf->record;
902  xl_heap_update *xlrec;
903  ReorderBufferChange *change;
904  char *data;
905  RelFileNode target_node;
906 
907  xlrec = (xl_heap_update *) XLogRecGetData(r);
908 
909  /* only interested in our database */
910  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
911  if (target_node.dbNode != ctx->slot->data.database)
912  return;
913 
914  /* output plugin doesn't look for this origin, no need to queue */
915  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
916  return;
917 
918  change = ReorderBufferGetChange(ctx->reorder);
920  change->origin_id = XLogRecGetOrigin(r);
921  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
922 
924  {
925  Size datalen;
926  Size tuplelen;
927 
928  data = XLogRecGetBlockData(r, 0, &datalen);
929 
930  tuplelen = datalen - SizeOfHeapHeader;
931 
932  change->data.tp.newtuple =
933  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
934 
935  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
936  }
937 
938  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
939  {
940  Size datalen;
941  Size tuplelen;
942 
943  /* caution, remaining data in record is not aligned */
945  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
946  tuplelen = datalen - SizeOfHeapHeader;
947 
948  change->data.tp.oldtuple =
949  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
950 
951  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
952  }
953 
954  change->data.tp.clear_toast_afterwards = true;
955 
957  change, false);
958 }
#define SizeOfHeapUpdate
Definition: heapam_xlog.h:227
#define XLH_UPDATE_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:84
#define XLH_UPDATE_CONTAINS_OLD
Definition: heapam_xlog.h:89
@ REORDER_BUFFER_CHANGE_UPDATE
Definition: reorderbuffer.h:57

References ReorderBufferChange::action, buf, ReorderBufferChange::data, ReplicationSlot::data, data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_update::flags, ReorderBufferChange::origin_id, 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 heap_decode().

◆ DecodeXLogTuple()

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

Definition at line 1200 of file decode.c.

1201 {
1202  xl_heap_header xlhdr;
1203  int datalen = len - SizeOfHeapHeader;
1205 
1206  Assert(datalen >= 0);
1207 
1208  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1209  header = tuple->tuple.t_data;
1210 
1211  /* not a disk based tuple */
1212  ItemPointerSetInvalid(&tuple->tuple.t_self);
1213 
1214  /* we can only figure this out after reassembling the transactions */
1215  tuple->tuple.t_tableOid = InvalidOid;
1216 
1217  /* data is not stored aligned, copy to aligned storage */
1218  memcpy((char *) &xlhdr,
1219  data,
1221 
1222  memset(header, 0, SizeofHeapTupleHeader);
1223 
1224  memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
1226  datalen);
1227 
1228  header->t_infomask = xlhdr.t_infomask;
1229  header->t_infomask2 = xlhdr.t_infomask2;
1230  header->t_hoff = xlhdr.t_hoff;
1231 }
const void size_t len
uint16 t_infomask
Definition: heapam_xlog.h:147
uint16 t_infomask2
Definition: heapam_xlog.h:146

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

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

◆ FilterByOrigin()

static bool FilterByOrigin ( LogicalDecodingContext ctx,
RepOriginId  origin_id 
)
inlinestatic

Definition at line 548 of file decode.c.

549 {
550  if (ctx->callbacks.filter_by_origin_cb == NULL)
551  return false;
552 
553  return filter_by_origin_cb_wrapper(ctx, origin_id);
554 }
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:1156
OutputPluginCallbacks callbacks
Definition: logical.h:53
LogicalDecodeFilterByOriginCB filter_by_origin_cb

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

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

◆ FilterPrepare()

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

Definition at line 525 of file decode.c.

527 {
528  /*
529  * Skip if decoding of two-phase transactions at PREPARE time is not
530  * enabled. In that case, all two-phase transactions are considered
531  * filtered out and will be applied as regular transactions at COMMIT
532  * PREPARED.
533  */
534  if (!ctx->twophase)
535  return true;
536 
537  /*
538  * The filter_prepare callback is optional. When not supplied, all
539  * prepared transactions should go through.
540  */
541  if (ctx->callbacks.filter_prepare_cb == NULL)
542  return false;
543 
544  return filter_prepare_cb_wrapper(ctx, xid, gid);
545 }
bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: logical.c:1124
LogicalDecodeFilterPrepareCB filter_prepare_cb

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

Referenced by xact_decode().

◆ heap2_decode()

void heap2_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 378 of file decode.c.

379 {
380  uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
381  TransactionId xid = XLogRecGetXid(buf->record);
382  SnapBuild *builder = ctx->snapshot_builder;
383 
384  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
385 
386  /*
387  * If we don't have snapshot or we are just fast-forwarding, there is no
388  * point in decoding changes.
389  */
391  ctx->fast_forward)
392  return;
393 
394  switch (info)
395  {
397  if (!ctx->fast_forward &&
398  SnapBuildProcessChange(builder, xid, buf->origptr))
399  DecodeMultiInsert(ctx, buf);
400  break;
401  case XLOG_HEAP2_NEW_CID:
402  {
403  xl_heap_new_cid *xlrec;
404 
405  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
406  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
407 
408  break;
409  }
410  case XLOG_HEAP2_REWRITE:
411 
412  /*
413  * Although these records only exist to serve the needs of logical
414  * decoding, all the work happens as part of crash or archive
415  * recovery, so we don't need to do anything here.
416  */
417  break;
418 
419  /*
420  * Everything else here is just low level physical stuff we're not
421  * interested in.
422  */
424  case XLOG_HEAP2_PRUNE:
425  case XLOG_HEAP2_VACUUM:
426  case XLOG_HEAP2_VISIBLE:
428  break;
429  default:
430  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
431  }
432 }
unsigned char uint8
Definition: c.h:439
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1058
#define ERROR
Definition: elog.h:33
#define elog(elevel,...)
Definition: elog.h:218
#define XLOG_HEAP2_PRUNE
Definition: heapam_xlog.h:54
#define XLOG_HEAP2_MULTI_INSERT
Definition: heapam_xlog.h:58
#define XLOG_HEAP2_VACUUM
Definition: heapam_xlog.h:55
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:53
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:59
#define XLOG_HEAP2_FREEZE_PAGE
Definition: heapam_xlog.h:56
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:60
#define XLOG_HEAP2_VISIBLE
Definition: heapam_xlog.h:57
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:722
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:772
@ SNAPBUILD_FULL_SNAPSHOT
Definition: snapbuild.h:39
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:408

References buf, DecodeMultiInsert(), elog, ERROR, LogicalDecodingContext::fast_forward, 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.

◆ heap_decode()

void heap_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 438 of file decode.c.

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

References buf, DecodeDelete(), DecodeInsert(), DecodeSpecConfirm(), DecodeTruncate(), DecodeUpdate(), elog, ERROR, LogicalDecodingContext::fast_forward, 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.

◆ LogicalDecodingProcessRecord()

void LogicalDecodingProcessRecord ( LogicalDecodingContext ctx,
XLogReaderState record 
)

Definition at line 91 of file decode.c.

92 {
94  TransactionId txid;
95  RmgrData rmgr;
96 
97  buf.origptr = ctx->reader->ReadRecPtr;
98  buf.endptr = ctx->reader->EndRecPtr;
99  buf.record = record;
100 
101  txid = XLogRecGetTopXid(record);
102 
103  /*
104  * If the top-level xid is valid, we need to assign the subxact to the
105  * top-level xact. We need to do this for all records, hence we do it
106  * before the switch.
107  */
108  if (TransactionIdIsValid(txid))
109  {
111  txid,
112  XLogRecGetXid(record),
113  buf.origptr);
114  }
115 
116  rmgr = GetRmgr(XLogRecGetRmid(record));
117 
118  if (rmgr.rm_decode != NULL)
119  rmgr.rm_decode(ctx, &buf);
120  else
121  {
122  /* just deal with xid, and done */
124  buf.origptr);
125  }
126 }
void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
XLogReaderState * reader
Definition: logical.h:42
void(* rm_decode)(struct LogicalDecodingContext *ctx, struct XLogRecordBuffer *buf)
XLogRecPtr EndRecPtr
Definition: xlogreader.h:207
XLogRecPtr ReadRecPtr
Definition: xlogreader.h:206
static RmgrData GetRmgr(RmgrId rmid)
#define XLogRecGetRmid(decoder)
Definition: xlogreader.h:409
#define XLogRecGetTopXid(decoder)
Definition: xlogreader.h:412

References buf, XLogReaderState::EndRecPtr, GetRmgr(), LogicalDecodingContext::reader, XLogReaderState::ReadRecPtr, LogicalDecodingContext::reorder, ReorderBufferAssignChild(), ReorderBufferProcessXid(), RmgrData::rm_decode, TransactionIdIsValid, XLogRecGetRmid, XLogRecGetTopXid, and XLogRecGetXid.

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

◆ logicalmsg_decode()

void logicalmsg_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 560 of file decode.c.

561 {
562  SnapBuild *builder = ctx->snapshot_builder;
563  XLogReaderState *r = buf->record;
564  TransactionId xid = XLogRecGetXid(r);
565  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
566  RepOriginId origin_id = XLogRecGetOrigin(r);
567  Snapshot snapshot;
568  xl_logical_message *message;
569 
570  if (info != XLOG_LOGICAL_MESSAGE)
571  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
572 
573  ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
574 
575  /*
576  * If we don't have snapshot or we are just fast-forwarding, there is no
577  * point in decoding messages.
578  */
580  ctx->fast_forward)
581  return;
582 
583  message = (xl_logical_message *) XLogRecGetData(r);
584 
585  if (message->dbId != ctx->slot->data.database ||
586  FilterByOrigin(ctx, origin_id))
587  return;
588 
589  if (message->transactional &&
590  !SnapBuildProcessChange(builder, xid, buf->origptr))
591  return;
592  else if (!message->transactional &&
594  SnapBuildXactNeedsSkip(builder, buf->origptr)))
595  return;
596 
597  snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
598  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
599  message->transactional,
600  message->message, /* first part of message is
601  * prefix */
602  message->message_size,
603  message->message + message->prefix_size);
604 }
#define XLOG_LOGICAL_MESSAGE
Definition: message.h:36
void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snapshot, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:662
bool transactional
Definition: message.h:23
char message[FLEXIBLE_ARRAY_MEMBER]
Definition: message.h:27
#define XLR_INFO_MASK
Definition: xlogrecord.h:62

References buf, ReplicationSlot::data, ReplicationSlotPersistentData::database, xl_logical_message::dbId, elog, ERROR, LogicalDecodingContext::fast_forward, FilterByOrigin(), xl_logical_message::message, xl_logical_message::message_size, xl_logical_message::prefix_size, LogicalDecodingContext::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.

◆ standby_decode()

void standby_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 332 of file decode.c.

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

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

◆ xact_decode()

void xact_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 175 of file decode.c.

176 {
177  SnapBuild *builder = ctx->snapshot_builder;
178  ReorderBuffer *reorder = ctx->reorder;
179  XLogReaderState *r = buf->record;
181 
182  /*
183  * If the snapshot isn't yet fully built, we cannot decode anything, so
184  * bail out.
185  */
187  return;
188 
189  switch (info)
190  {
191  case XLOG_XACT_COMMIT:
193  {
194  xl_xact_commit *xlrec;
195  xl_xact_parsed_commit parsed;
196  TransactionId xid;
197  bool two_phase = false;
198 
199  xlrec = (xl_xact_commit *) XLogRecGetData(r);
200  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
201 
202  if (!TransactionIdIsValid(parsed.twophase_xid))
203  xid = XLogRecGetXid(r);
204  else
205  xid = parsed.twophase_xid;
206 
207  /*
208  * We would like to process the transaction in a two-phase
209  * manner iff output plugin supports two-phase commits and
210  * doesn't filter the transaction at prepare time.
211  */
212  if (info == XLOG_XACT_COMMIT_PREPARED)
213  two_phase = !(FilterPrepare(ctx, xid,
214  parsed.twophase_gid));
215 
216  DecodeCommit(ctx, buf, &parsed, xid, two_phase);
217  break;
218  }
219  case XLOG_XACT_ABORT:
221  {
222  xl_xact_abort *xlrec;
223  xl_xact_parsed_abort parsed;
224  TransactionId xid;
225  bool two_phase = false;
226 
227  xlrec = (xl_xact_abort *) XLogRecGetData(r);
228  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
229 
230  if (!TransactionIdIsValid(parsed.twophase_xid))
231  xid = XLogRecGetXid(r);
232  else
233  xid = parsed.twophase_xid;
234 
235  /*
236  * We would like to process the transaction in a two-phase
237  * manner iff output plugin supports two-phase commits and
238  * doesn't filter the transaction at prepare time.
239  */
240  if (info == XLOG_XACT_ABORT_PREPARED)
241  two_phase = !(FilterPrepare(ctx, xid,
242  parsed.twophase_gid));
243 
244  DecodeAbort(ctx, buf, &parsed, xid, two_phase);
245  break;
246  }
248 
249  /*
250  * We assign subxact to the toplevel xact while processing each
251  * record if required. So, we don't need to do anything here. See
252  * LogicalDecodingProcessRecord.
253  */
254  break;
256  {
257  TransactionId xid;
258  xl_xact_invals *invals;
259 
260  xid = XLogRecGetXid(r);
261  invals = (xl_xact_invals *) XLogRecGetData(r);
262 
263  /*
264  * Execute the invalidations for xid-less transactions,
265  * otherwise, accumulate them so that they can be processed at
266  * the commit time.
267  */
268  if (TransactionIdIsValid(xid))
269  {
270  if (!ctx->fast_forward)
271  ReorderBufferAddInvalidations(reorder, xid,
272  buf->origptr,
273  invals->nmsgs,
274  invals->msgs);
276  buf->origptr);
277  }
278  else if ((!ctx->fast_forward))
280  invals->nmsgs,
281  invals->msgs);
282  }
283  break;
284  case XLOG_XACT_PREPARE:
285  {
286  xl_xact_parsed_prepare parsed;
287  xl_xact_prepare *xlrec;
288 
289  /* ok, parse it */
290  xlrec = (xl_xact_prepare *) XLogRecGetData(r);
292  xlrec, &parsed);
293 
294  /*
295  * We would like to process the transaction in a two-phase
296  * manner iff output plugin supports two-phase commits and
297  * doesn't filter the transaction at prepare time.
298  */
299  if (FilterPrepare(ctx, parsed.twophase_xid,
300  parsed.twophase_gid))
301  {
302  ReorderBufferProcessXid(reorder, parsed.twophase_xid,
303  buf->origptr);
304  break;
305  }
306 
307  /*
308  * Note that if the prepared transaction has locked [user]
309  * catalog tables exclusively then decoding prepare can block
310  * till the main transaction is committed because it needs to
311  * lock the catalog tables.
312  *
313  * XXX Now, this can even lead to a deadlock if the prepare
314  * transaction is waiting to get it logically replicated for
315  * distributed 2PC. This can be avoided by disallowing
316  * preparing transactions that have locked [user] catalog
317  * tables exclusively but as of now, we ask users not to do
318  * such an operation.
319  */
320  DecodePrepare(ctx, buf, &parsed);
321  break;
322  }
323  default:
324  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
325  }
326 }
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:785
static bool FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: decode.c:525
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:615
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
Definition: decode.c:710
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER]
Definition: xact.h:286
int nmsgs
Definition: xact.h:285
TransactionId twophase_xid
Definition: xact.h:408
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:159
#define XLOG_XACT_INVALIDATIONS
Definition: xact.h:162
#define XLOG_XACT_PREPARE
Definition: xact.h:157
#define XLOG_XACT_COMMIT
Definition: xact.h:156
#define XLOG_XACT_OPMASK
Definition: xact.h:166
#define XLOG_XACT_ABORT
Definition: xact.h:158
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:161
#define XLOG_XACT_ABORT_PREPARED
Definition: xact.h:160
void ParseCommitRecord(uint8 info, xl_xact_commit *xlrec, xl_xact_parsed_commit *parsed)
Definition: xactdesc.c:35
void ParseAbortRecord(uint8 info, xl_xact_abort *xlrec, xl_xact_parsed_abort *parsed)
Definition: xactdesc.c:141
void ParsePrepareRecord(uint8 info, xl_xact_prepare *xlrec, xl_xact_parsed_prepare *parsed)
Definition: xactdesc.c:239

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

◆ xlog_decode()

void xlog_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 132 of file decode.c.

133 {
134  SnapBuild *builder = ctx->snapshot_builder;
135  uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
136 
138  buf->origptr);
139 
140  switch (info)
141  {
142  /* this is also used in END_OF_RECOVERY checkpoints */
145  SnapBuildSerializationPoint(builder, buf->origptr);
146 
147  break;
149 
150  /*
151  * a RUNNING_XACTS record will have been logged near to this, we
152  * can restart from there.
153  */
154  break;
155  case XLOG_NOOP:
156  case XLOG_NEXTOID:
157  case XLOG_SWITCH:
158  case XLOG_BACKUP_END:
160  case XLOG_RESTORE_POINT:
161  case XLOG_FPW_CHANGE:
162  case XLOG_FPI_FOR_HINT:
163  case XLOG_FPI:
165  break;
166  default:
167  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
168  }
169 }
#define XLOG_RESTORE_POINT
Definition: pg_control.h:74
#define XLOG_FPW_CHANGE
Definition: pg_control.h:75
#define XLOG_OVERWRITE_CONTRECORD
Definition: pg_control.h:80
#define XLOG_FPI
Definition: pg_control.h:78
#define XLOG_FPI_FOR_HINT
Definition: pg_control.h:77
#define XLOG_NEXTOID
Definition: pg_control.h:70
#define XLOG_NOOP
Definition: pg_control.h:69
#define XLOG_CHECKPOINT_SHUTDOWN
Definition: pg_control.h:67
#define XLOG_SWITCH
Definition: pg_control.h:71
#define XLOG_BACKUP_END
Definition: pg_control.h:72
#define XLOG_PARAMETER_CHANGE
Definition: pg_control.h:73
#define XLOG_CHECKPOINT_ONLINE
Definition: pg_control.h:68
#define XLOG_END_OF_RECOVERY
Definition: pg_control.h:76
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1479

References buf, elog, ERROR, 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_OVERWRITE_CONTRECORD, XLOG_PARAMETER_CHANGE, XLOG_RESTORE_POINT, XLOG_SWITCH, XLogRecGetInfo, XLogRecGetXid, and XLR_INFO_MASK.