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, HeapTuple tuple)
 
static bool FilterPrepare (LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
 
static bool DecodeTXNNeedSkip (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid txn_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 840 of file decode.c.

843 {
844  int i;
845  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
846  TimestampTz abort_time = parsed->xact_time;
847  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
848  bool skip_xact;
849 
850  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
851  {
852  origin_lsn = parsed->origin_lsn;
853  abort_time = parsed->origin_timestamp;
854  }
855 
856  /*
857  * Check whether we need to process this transaction. See
858  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
859  * transaction.
860  */
861  skip_xact = DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id);
862 
863  /*
864  * Send the final rollback record for a prepared transaction unless we
865  * need to skip it. For non-two-phase xacts, simply forget the xact.
866  */
867  if (two_phase && !skip_xact)
868  {
869  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
871  abort_time, origin_id, origin_lsn,
872  parsed->twophase_gid, false);
873  }
874  else
875  {
876  for (i = 0; i < parsed->nsubxacts; i++)
877  {
878  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
879  buf->record->EndRecPtr, abort_time);
880  }
881 
882  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr,
883  abort_time);
884  }
885 
886  /* update the decoding stats */
887  UpdateDecodingStats(ctx);
888 }
int64 TimestampTz
Definition: timestamp.h:39
static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid txn_dbid, RepOriginId origin_id)
Definition: decode.c:1302
int i
Definition: isn.c:73
void UpdateDecodingStats(LogicalDecodingContext *ctx)
Definition: logical.c:1926
static bool two_phase
static char * buf
Definition: pg_test_fsync.c:73
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, TimestampTz abort_time)
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)
struct ReorderBuffer * reorder
Definition: logical.h:43
TimestampTz xact_time
Definition: xact.h:406
TransactionId * subxacts
Definition: xact.h:413
XLogRecPtr origin_lsn
Definition: xact.h:424
char twophase_gid[GIDSIZE]
Definition: xact.h:422
TimestampTz origin_timestamp
Definition: xact.h:425
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:193
uint16 RepOriginId
Definition: xlogdefs.h:65
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:413

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

672 {
673  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
674  TimestampTz commit_time = parsed->xact_time;
675  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
676  int i;
677 
678  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
679  {
680  origin_lsn = parsed->origin_lsn;
681  commit_time = parsed->origin_timestamp;
682  }
683 
684  SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
685  parsed->nsubxacts, parsed->subxacts,
686  parsed->xinfo);
687 
688  /* ----
689  * Check whether we are interested in this specific transaction, and tell
690  * the reorderbuffer to forget the content of the (sub-)transactions
691  * if not.
692  *
693  * We can't just use ReorderBufferAbort() here, because we need to execute
694  * the transaction's invalidations. This currently won't be needed if
695  * we're just skipping over the transaction because currently we only do
696  * so during startup, to get to the first transaction the client needs. As
697  * we have reset the catalog caches before starting to read WAL, and we
698  * haven't yet touched any catalogs, there can't be anything to invalidate.
699  * But if we're "forgetting" this commit because it happened in another
700  * database, the invalidations might be important, because they could be
701  * for shared catalogs and we might have loaded data into the relevant
702  * syscaches.
703  * ---
704  */
705  if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
706  {
707  for (i = 0; i < parsed->nsubxacts; i++)
708  {
709  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
710  }
711  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
712 
713  return;
714  }
715 
716  /* tell the reorderbuffer about the surviving subtransactions */
717  for (i = 0; i < parsed->nsubxacts; i++)
718  {
719  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
720  buf->origptr, buf->endptr);
721  }
722 
723  /*
724  * Send the final commit record if the transaction data is already
725  * decoded, otherwise, process the entire transaction.
726  */
727  if (two_phase)
728  {
729  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
731  commit_time, origin_id, origin_lsn,
732  parsed->twophase_gid, true);
733  }
734  else
735  {
736  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
737  commit_time, origin_id, origin_lsn);
738  }
739 
740  /*
741  * Update the decoding stats at transaction prepare/commit/abort.
742  * Additionally we send the stats when we spill or stream the changes to
743  * avoid losing them in case the decoding is interrupted. It is not clear
744  * that sending more or less frequently than this would be better.
745  */
746  UpdateDecodingStats(ctx);
747 }
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:416
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts, uint32 xinfo)
Definition: snapbuild.c:1025
struct SnapBuild * snapshot_builder
Definition: logical.h:44
TimestampTz xact_time
Definition: xact.h:373
TimestampTz origin_timestamp
Definition: xact.h:399
TransactionId * subxacts
Definition: xact.h:380
char twophase_gid[GIDSIZE]
Definition: xact.h:392
XLogRecPtr origin_lsn
Definition: xact.h:398

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

1023 {
1024  XLogReaderState *r = buf->record;
1025  xl_heap_delete *xlrec;
1026  ReorderBufferChange *change;
1027  RelFileLocator target_locator;
1028 
1029  xlrec = (xl_heap_delete *) XLogRecGetData(r);
1030 
1031  /* only interested in our database */
1032  XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
1033  if (target_locator.dbOid != ctx->slot->data.database)
1034  return;
1035 
1036  /* output plugin doesn't look for this origin, no need to queue */
1037  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1038  return;
1039 
1040  change = ReorderBufferGetChange(ctx->reorder);
1041 
1042  if (xlrec->flags & XLH_DELETE_IS_SUPER)
1044  else
1046 
1047  change->origin_id = XLogRecGetOrigin(r);
1048 
1049  memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
1050 
1051  /* old primary key stored */
1052  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
1053  {
1054  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
1055  Size tuplelen = datalen - SizeOfHeapHeader;
1056 
1058 
1059  change->data.tp.oldtuple =
1060  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1061 
1062  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
1063  datalen, change->data.tp.oldtuple);
1064  }
1065 
1066  change->data.tp.clear_toast_afterwards = true;
1067 
1069  change, false);
1070 }
size_t Size
Definition: c.h:594
static void DecodeXLogTuple(char *data, Size len, HeapTuple tuple)
Definition: decode.c:1256
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:576
#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')
HeapTuple ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT
Definition: reorderbuffer.h:56
@ REORDER_BUFFER_CHANGE_DELETE
Definition: reorderbuffer.h:48
ReplicationSlot * slot
Definition: logical.h:39
struct ReorderBufferChange::@100::@101 tp
ReorderBufferChangeType action
Definition: reorderbuffer.h:75
RepOriginId origin_id
Definition: reorderbuffer.h:80
union ReorderBufferChange::@100 data
ReplicationSlotPersistentData data
Definition: slot.h:181
void XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileLocator *rlocator, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1973
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:416
#define XLogRecGetData(decoder)
Definition: xlogreader.h:415
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:412

References ReorderBufferChange::action, Assert(), buf, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileLocator::dbOid, 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 896 of file decode.c.

897 {
898  Size datalen;
899  char *tupledata;
900  Size tuplelen;
901  XLogReaderState *r = buf->record;
902  xl_heap_insert *xlrec;
903  ReorderBufferChange *change;
904  RelFileLocator target_locator;
905 
906  xlrec = (xl_heap_insert *) XLogRecGetData(r);
907 
908  /*
909  * Ignore insert records without new tuples (this does happen when
910  * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
911  */
912  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
913  return;
914 
915  /* only interested in our database */
916  XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
917  if (target_locator.dbOid != ctx->slot->data.database)
918  return;
919 
920  /* output plugin doesn't look for this origin, no need to queue */
921  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
922  return;
923 
924  change = ReorderBufferGetChange(ctx->reorder);
925  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
927  else
929  change->origin_id = XLogRecGetOrigin(r);
930 
931  memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
932 
933  tupledata = XLogRecGetBlockData(r, 0, &datalen);
934  tuplelen = datalen - SizeOfHeapHeader;
935 
936  change->data.tp.newtuple =
937  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
938 
939  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
940 
941  change->data.tp.clear_toast_afterwards = true;
942 
944  change,
946 }
#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:46
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT
Definition: reorderbuffer.h:54
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:2027

References ReorderBufferChange::action, buf, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileLocator::dbOid, 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 1114 of file decode.c.

1115 {
1116  XLogReaderState *r = buf->record;
1117  xl_heap_multi_insert *xlrec;
1118  int i;
1119  char *data;
1120  char *tupledata;
1121  Size tuplelen;
1122  RelFileLocator rlocator;
1123 
1124  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
1125 
1126  /*
1127  * Ignore insert records without new tuples. This happens when a
1128  * multi_insert is done on a catalog or on a non-persistent relation.
1129  */
1130  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
1131  return;
1132 
1133  /* only interested in our database */
1134  XLogRecGetBlockTag(r, 0, &rlocator, NULL, NULL);
1135  if (rlocator.dbOid != ctx->slot->data.database)
1136  return;
1137 
1138  /* output plugin doesn't look for this origin, no need to queue */
1139  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1140  return;
1141 
1142  /*
1143  * We know that this multi_insert isn't for a catalog, so the block should
1144  * always have data even if a full-page write of it is taken.
1145  */
1146  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
1147  Assert(tupledata != NULL);
1148 
1149  data = tupledata;
1150  for (i = 0; i < xlrec->ntuples; i++)
1151  {
1152  ReorderBufferChange *change;
1153  xl_multi_insert_tuple *xlhdr;
1154  int datalen;
1155  HeapTuple tuple;
1156  HeapTupleHeader header;
1157 
1158  change = ReorderBufferGetChange(ctx->reorder);
1160  change->origin_id = XLogRecGetOrigin(r);
1161 
1162  memcpy(&change->data.tp.rlocator, &rlocator, sizeof(RelFileLocator));
1163 
1164  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
1165  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
1166  datalen = xlhdr->datalen;
1167 
1168  change->data.tp.newtuple =
1169  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
1170 
1171  tuple = change->data.tp.newtuple;
1172  header = tuple->t_data;
1173 
1174  /* not a disk based tuple */
1175  ItemPointerSetInvalid(&tuple->t_self);
1176 
1177  /*
1178  * We can only figure this out after reassembling the transactions.
1179  */
1180  tuple->t_tableOid = InvalidOid;
1181 
1182  tuple->t_len = datalen + SizeofHeapTupleHeader;
1183 
1184  memset(header, 0, SizeofHeapTupleHeader);
1185 
1186  memcpy((char *) tuple->t_data + SizeofHeapTupleHeader,
1187  (char *) data,
1188  datalen);
1189  header->t_infomask = xlhdr->t_infomask;
1190  header->t_infomask2 = xlhdr->t_infomask2;
1191  header->t_hoff = xlhdr->t_hoff;
1192 
1193  /*
1194  * Reset toast reassembly state only after the last row in the last
1195  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
1196  * call.
1197  */
1198  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
1199  (i + 1) == xlrec->ntuples)
1200  change->data.tp.clear_toast_afterwards = true;
1201  else
1202  change->data.tp.clear_toast_afterwards = false;
1203 
1205  buf->origptr, change, false);
1206 
1207  /* move to the next xl_multi_insert_tuple entry */
1208  data += datalen;
1209  }
1210  Assert(data == tupledata + tuplelen);
1211 }
#define SHORTALIGN(LEN)
Definition: c.h:796
#define XLH_INSERT_LAST_IN_MULTI
Definition: heapam_xlog.h:67
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:193
#define SizeofHeapTupleHeader
Definition: htup_details.h:185
static void ItemPointerSetInvalid(ItemPointerData *pointer)
Definition: itemptr.h:184
const void * data
#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

References ReorderBufferChange::action, Assert(), buf, ReorderBufferChange::data, ReplicationSlot::data, data, ReplicationSlotPersistentData::database, xl_multi_insert_tuple::datalen, RelFileLocator::dbOid, FilterByOrigin(), xl_heap_multi_insert::flags, 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, HeapTupleHeaderData::t_hoff, xl_multi_insert_tuple::t_infomask, HeapTupleHeaderData::t_infomask, xl_multi_insert_tuple::t_infomask2, HeapTupleHeaderData::t_infomask2, HeapTupleData::t_len, HeapTupleData::t_self, HeapTupleData::t_tableOid, ReorderBufferChange::tp, 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 765 of file decode.c.

767 {
768  SnapBuild *builder = ctx->snapshot_builder;
769  XLogRecPtr origin_lsn = parsed->origin_lsn;
770  TimestampTz prepare_time = parsed->xact_time;
771  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
772  int i;
773  TransactionId xid = parsed->twophase_xid;
774 
775  if (parsed->origin_timestamp != 0)
776  prepare_time = parsed->origin_timestamp;
777 
778  /*
779  * Remember the prepare info for a txn so that it can be used later in
780  * commit prepared if required. See ReorderBufferFinishPrepared.
781  */
782  if (!ReorderBufferRememberPrepareInfo(ctx->reorder, xid, buf->origptr,
783  buf->endptr, prepare_time, origin_id,
784  origin_lsn))
785  return;
786 
787  /* We can't start streaming unless a consistent state is reached. */
789  {
791  return;
792  }
793 
794  /*
795  * Check whether we need to process this transaction. See
796  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
797  * transaction.
798  *
799  * We can't call ReorderBufferForget as we did in DecodeCommit as the txn
800  * hasn't yet been committed, removing this txn before a commit might
801  * result in the computation of an incorrect restart_lsn. See
802  * SnapBuildProcessRunningXacts. But we need to process cache
803  * invalidations if there are any for the reasons mentioned in
804  * DecodeCommit.
805  */
806  if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
807  {
809  ReorderBufferInvalidate(ctx->reorder, xid, buf->origptr);
810  return;
811  }
812 
813  /* Tell the reorderbuffer about the surviving subtransactions. */
814  for (i = 0; i < parsed->nsubxacts; i++)
815  {
816  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
817  buf->origptr, buf->endptr);
818  }
819 
820  /* replay actions of all transaction + subtransactions in order */
821  ReorderBufferPrepare(ctx->reorder, xid, parsed->twophase_gid);
822 
823  /*
824  * Update the decoding stats at transaction prepare/commit/abort.
825  * Additionally we send the stats when we spill or stream the changes to
826  * avoid losing them in case the decoding is interrupted. It is not clear
827  * that sending more or less frequently than this would be better.
828  */
829  UpdateDecodingStats(ctx);
830 }
uint32 TransactionId
Definition: c.h:641
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:407
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:46
TransactionId twophase_xid
Definition: xact.h:391

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

1221 {
1222  XLogReaderState *r = buf->record;
1223  ReorderBufferChange *change;
1224  RelFileLocator target_locator;
1225 
1226  /* only interested in our database */
1227  XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
1228  if (target_locator.dbOid != ctx->slot->data.database)
1229  return;
1230 
1231  /* output plugin doesn't look for this origin, no need to queue */
1232  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1233  return;
1234 
1235  change = ReorderBufferGetChange(ctx->reorder);
1237  change->origin_id = XLogRecGetOrigin(r);
1238 
1239  memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
1240 
1241  change->data.tp.clear_toast_afterwards = true;
1242 
1244  change, false);
1245 }
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM
Definition: reorderbuffer.h:55

References ReorderBufferChange::action, buf, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileLocator::dbOid, 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 1076 of file decode.c.

1077 {
1078  XLogReaderState *r = buf->record;
1079  xl_heap_truncate *xlrec;
1080  ReorderBufferChange *change;
1081 
1082  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
1083 
1084  /* only interested in our database */
1085  if (xlrec->dbId != ctx->slot->data.database)
1086  return;
1087 
1088  /* output plugin doesn't look for this origin, no need to queue */
1089  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1090  return;
1091 
1092  change = ReorderBufferGetChange(ctx->reorder);
1094  change->origin_id = XLogRecGetOrigin(r);
1095  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
1096  change->data.truncate.cascade = true;
1097  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
1098  change->data.truncate.restart_seqs = true;
1099  change->data.truncate.nrelids = xlrec->nrelids;
1100  change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
1101  xlrec->nrelids);
1102  memcpy(change->data.truncate.relids, xlrec->relids,
1103  xlrec->nrelids * sizeof(Oid));
1105  buf->origptr, change, false);
1106 }
#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:57
struct ReorderBufferChange::@100::@102 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  txn_dbid,
RepOriginId  origin_id 
)
static

Definition at line 1302 of file decode.c.

1304 {
1305  if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
1306  (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) ||
1307  FilterByOrigin(ctx, origin_id))
1308  return true;
1309 
1310  /*
1311  * We also skip decoding in fast_forward mode. In passing set the
1312  * processing_required flag to indicate that if it were not for
1313  * fast_forward mode, processing would have been required.
1314  */
1315  if (ctx->fast_forward)
1316  {
1317  ctx->processing_required = true;
1318  return true;
1319  }
1320 
1321  return false;
1322 }
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:434

References buf, ReplicationSlot::data, ReplicationSlotPersistentData::database, LogicalDecodingContext::fast_forward, FilterByOrigin(), InvalidOid, LogicalDecodingContext::processing_required, 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 955 of file decode.c.

956 {
957  XLogReaderState *r = buf->record;
958  xl_heap_update *xlrec;
959  ReorderBufferChange *change;
960  char *data;
961  RelFileLocator target_locator;
962 
963  xlrec = (xl_heap_update *) XLogRecGetData(r);
964 
965  /* only interested in our database */
966  XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
967  if (target_locator.dbOid != ctx->slot->data.database)
968  return;
969 
970  /* output plugin doesn't look for this origin, no need to queue */
971  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
972  return;
973 
974  change = ReorderBufferGetChange(ctx->reorder);
976  change->origin_id = XLogRecGetOrigin(r);
977  memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
978 
980  {
981  Size datalen;
982  Size tuplelen;
983 
984  data = XLogRecGetBlockData(r, 0, &datalen);
985 
986  tuplelen = datalen - SizeOfHeapHeader;
987 
988  change->data.tp.newtuple =
989  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
990 
991  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
992  }
993 
994  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
995  {
996  Size datalen;
997  Size tuplelen;
998 
999  /* caution, remaining data in record is not aligned */
1001  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
1002  tuplelen = datalen - SizeOfHeapHeader;
1003 
1004  change->data.tp.oldtuple =
1005  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1006 
1007  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
1008  }
1009 
1010  change->data.tp.clear_toast_afterwards = true;
1011 
1013  change, false);
1014 }
#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:47

References ReorderBufferChange::action, buf, ReorderBufferChange::data, ReplicationSlot::data, data, ReplicationSlotPersistentData::database, RelFileLocator::dbOid, 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,
HeapTuple  tuple 
)
static

Definition at line 1256 of file decode.c.

1257 {
1258  xl_heap_header xlhdr;
1259  int datalen = len - SizeOfHeapHeader;
1260  HeapTupleHeader header;
1261 
1262  Assert(datalen >= 0);
1263 
1264  tuple->t_len = datalen + SizeofHeapTupleHeader;
1265  header = tuple->t_data;
1266 
1267  /* not a disk based tuple */
1268  ItemPointerSetInvalid(&tuple->t_self);
1269 
1270  /* we can only figure this out after reassembling the transactions */
1271  tuple->t_tableOid = InvalidOid;
1272 
1273  /* data is not stored aligned, copy to aligned storage */
1274  memcpy((char *) &xlhdr,
1275  data,
1277 
1278  memset(header, 0, SizeofHeapTupleHeader);
1279 
1280  memcpy(((char *) tuple->t_data) + SizeofHeapTupleHeader,
1282  datalen);
1283 
1284  header->t_infomask = xlhdr.t_infomask;
1285  header->t_infomask2 = xlhdr.t_infomask2;
1286  header->t_hoff = xlhdr.t_hoff;
1287 }
const void size_t len
uint16 t_infomask
Definition: heapam_xlog.h:147
uint16 t_infomask2
Definition: heapam_xlog.h:146

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

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

◆ FilterByOrigin()

static bool FilterByOrigin ( LogicalDecodingContext ctx,
RepOriginId  origin_id 
)
inlinestatic

Definition at line 576 of file decode.c.

577 {
578  if (ctx->callbacks.filter_by_origin_cb == NULL)
579  return false;
580 
581  return filter_by_origin_cb_wrapper(ctx, origin_id);
582 }
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:1210
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 553 of file decode.c.

555 {
556  /*
557  * Skip if decoding of two-phase transactions at PREPARE time is not
558  * enabled. In that case, all two-phase transactions are considered
559  * filtered out and will be applied as regular transactions at COMMIT
560  * PREPARED.
561  */
562  if (!ctx->twophase)
563  return true;
564 
565  /*
566  * The filter_prepare callback is optional. When not supplied, all
567  * prepared transactions should go through.
568  */
569  if (ctx->callbacks.filter_prepare_cb == NULL)
570  return false;
571 
572  return filter_prepare_cb_wrapper(ctx, xid, gid);
573 }
bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: logical.c:1178
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 407 of file decode.c.

408 {
409  uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
410  TransactionId xid = XLogRecGetXid(buf->record);
411  SnapBuild *builder = ctx->snapshot_builder;
412 
413  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
414 
415  /*
416  * If we don't have snapshot or we are just fast-forwarding, there is no
417  * point in decoding changes.
418  */
420  ctx->fast_forward)
421  return;
422 
423  switch (info)
424  {
426  if (SnapBuildProcessChange(builder, xid, buf->origptr))
427  DecodeMultiInsert(ctx, buf);
428  break;
429  case XLOG_HEAP2_NEW_CID:
430  {
431  xl_heap_new_cid *xlrec;
432 
433  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
434  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
435 
436  break;
437  }
438  case XLOG_HEAP2_REWRITE:
439 
440  /*
441  * Although these records only exist to serve the needs of logical
442  * decoding, all the work happens as part of crash or archive
443  * recovery, so we don't need to do anything here.
444  */
445  break;
446 
447  /*
448  * Everything else here is just low level physical stuff we're not
449  * interested in.
450  */
452  case XLOG_HEAP2_PRUNE:
453  case XLOG_HEAP2_VACUUM:
454  case XLOG_HEAP2_VISIBLE:
456  break;
457  default:
458  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
459  }
460 }
unsigned char uint8
Definition: c.h:493
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1114
#define ERROR
Definition: elog.h:39
#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:769
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:819
@ SNAPBUILD_FULL_SNAPSHOT
Definition: snapbuild.h:39
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:410

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

467 {
468  uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
469  TransactionId xid = XLogRecGetXid(buf->record);
470  SnapBuild *builder = ctx->snapshot_builder;
471 
472  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
473 
474  /*
475  * If we don't have snapshot or we are just fast-forwarding, there is no
476  * point in decoding data changes.
477  */
479  ctx->fast_forward)
480  return;
481 
482  switch (info)
483  {
484  case XLOG_HEAP_INSERT:
485  if (SnapBuildProcessChange(builder, xid, buf->origptr))
486  DecodeInsert(ctx, buf);
487  break;
488 
489  /*
490  * Treat HOT update as normal updates. There is no useful
491  * information in the fact that we could make it a HOT update
492  * locally and the WAL layout is compatible.
493  */
495  case XLOG_HEAP_UPDATE:
496  if (SnapBuildProcessChange(builder, xid, buf->origptr))
497  DecodeUpdate(ctx, buf);
498  break;
499 
500  case XLOG_HEAP_DELETE:
501  if (SnapBuildProcessChange(builder, xid, buf->origptr))
502  DecodeDelete(ctx, buf);
503  break;
504 
505  case XLOG_HEAP_TRUNCATE:
506  if (SnapBuildProcessChange(builder, xid, buf->origptr))
507  DecodeTruncate(ctx, buf);
508  break;
509 
510  case XLOG_HEAP_INPLACE:
511 
512  /*
513  * Inplace updates are only ever performed on catalog tuples and
514  * can, per definition, not change tuple visibility. Since we
515  * don't decode catalog tuples, we're not interested in the
516  * record's contents.
517  *
518  * In-place updates can be used either by XID-bearing transactions
519  * (e.g. in CREATE INDEX CONCURRENTLY) or by XID-less
520  * transactions (e.g. VACUUM). In the former case, the commit
521  * record will include cache invalidations, so we mark the
522  * transaction as catalog modifying here. Currently that's
523  * redundant because the commit will do that as well, but once we
524  * support decoding in-progress relations, this will be important.
525  */
526  if (!TransactionIdIsValid(xid))
527  break;
528 
529  (void) SnapBuildProcessChange(builder, xid, buf->origptr);
530  ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr);
531  break;
532 
533  case XLOG_HEAP_CONFIRM:
534  if (SnapBuildProcessChange(builder, xid, buf->origptr))
535  DecodeSpecConfirm(ctx, buf);
536  break;
537 
538  case XLOG_HEAP_LOCK:
539  /* we don't care about row level locks for now */
540  break;
541 
542  default:
543  elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
544  break;
545  }
546 }
static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1022
static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:896
static void DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1076
static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:955
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1220
#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:411
#define XLogRecGetTopXid(decoder)
Definition: xlogreader.h:414

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

Referenced by DecodingContextFindStartpoint(), LogicalReplicationSlotHasPendingWal(), 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 588 of file decode.c.

589 {
590  SnapBuild *builder = ctx->snapshot_builder;
591  XLogReaderState *r = buf->record;
592  TransactionId xid = XLogRecGetXid(r);
593  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
594  RepOriginId origin_id = XLogRecGetOrigin(r);
595  Snapshot snapshot = NULL;
596  xl_logical_message *message;
597 
598  if (info != XLOG_LOGICAL_MESSAGE)
599  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
600 
601  ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
602 
603  /* If we don't have snapshot, there is no point in decoding messages */
605  return;
606 
607  message = (xl_logical_message *) XLogRecGetData(r);
608 
609  if (message->dbId != ctx->slot->data.database ||
610  FilterByOrigin(ctx, origin_id))
611  return;
612 
613  if (message->transactional &&
614  !SnapBuildProcessChange(builder, xid, buf->origptr))
615  return;
616  else if (!message->transactional &&
618  SnapBuildXactNeedsSkip(builder, buf->origptr)))
619  return;
620 
621  /*
622  * We also skip decoding in fast_forward mode. This check must be last
623  * because we don't want to set the processing_required flag unless we
624  * have a decodable message.
625  */
626  if (ctx->fast_forward)
627  {
628  /*
629  * We need to set processing_required flag to notify the message's
630  * existence to the caller. Usually, the flag is set when either the
631  * COMMIT or ABORT records are decoded, but this must be turned on
632  * here because the non-transactional logical message is decoded
633  * without waiting for these records.
634  */
635  if (!message->transactional)
636  ctx->processing_required = true;
637 
638  return;
639  }
640 
641  /*
642  * If this is a non-transactional change, get the snapshot we're expected
643  * to use. We only get here when the snapshot is consistent, and the
644  * change is not meant to be skipped.
645  *
646  * For transactional changes we don't need a snapshot, we'll use the
647  * regular snapshot maintained by ReorderBuffer. We just leave it NULL.
648  */
649  if (!message->transactional)
650  snapshot = SnapBuildGetOrBuildSnapshot(builder);
651 
652  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
653  message->transactional,
654  message->message, /* first part of message is
655  * prefix */
656  message->message_size,
657  message->message + message->prefix_size);
658 }
#define XLOG_LOGICAL_MESSAGE
Definition: message.h:37
void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snap, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:709
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::processing_required, 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 361 of file decode.c.

362 {
363  SnapBuild *builder = ctx->snapshot_builder;
364  XLogReaderState *r = buf->record;
365  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
366 
367  ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
368 
369  switch (info)
370  {
371  case XLOG_RUNNING_XACTS:
372  {
374 
375  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
376 
377  /*
378  * Abort all transactions that we keep track of, that are
379  * older than the record's oldestRunningXid. This is the most
380  * convenient spot for doing so since, in contrast to shutdown
381  * or end-of-recovery checkpoints, we have information about
382  * all running transactions which includes prepared ones,
383  * while shutdown checkpoints just know that no non-prepared
384  * transactions are in progress.
385  */
387  }
388  break;
389  case XLOG_STANDBY_LOCK:
390  break;
391  case XLOG_INVALIDATIONS:
392 
393  /*
394  * We are processing the invalidations at the command level via
395  * XLOG_XACT_INVALIDATIONS. So we don't need to do anything here.
396  */
397  break;
398  default:
399  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
400  }
401 }
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1218
#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 204 of file decode.c.

205 {
206  SnapBuild *builder = ctx->snapshot_builder;
207  ReorderBuffer *reorder = ctx->reorder;
208  XLogReaderState *r = buf->record;
210 
211  /*
212  * If the snapshot isn't yet fully built, we cannot decode anything, so
213  * bail out.
214  */
216  return;
217 
218  switch (info)
219  {
220  case XLOG_XACT_COMMIT:
222  {
223  xl_xact_commit *xlrec;
224  xl_xact_parsed_commit parsed;
225  TransactionId xid;
226  bool two_phase = false;
227 
228  xlrec = (xl_xact_commit *) XLogRecGetData(r);
229  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
230 
231  if (!TransactionIdIsValid(parsed.twophase_xid))
232  xid = XLogRecGetXid(r);
233  else
234  xid = parsed.twophase_xid;
235 
236  /*
237  * We would like to process the transaction in a two-phase
238  * manner iff output plugin supports two-phase commits and
239  * doesn't filter the transaction at prepare time.
240  */
241  if (info == XLOG_XACT_COMMIT_PREPARED)
242  two_phase = !(FilterPrepare(ctx, xid,
243  parsed.twophase_gid));
244 
245  DecodeCommit(ctx, buf, &parsed, xid, two_phase);
246  break;
247  }
248  case XLOG_XACT_ABORT:
250  {
251  xl_xact_abort *xlrec;
252  xl_xact_parsed_abort parsed;
253  TransactionId xid;
254  bool two_phase = false;
255 
256  xlrec = (xl_xact_abort *) XLogRecGetData(r);
257  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
258 
259  if (!TransactionIdIsValid(parsed.twophase_xid))
260  xid = XLogRecGetXid(r);
261  else
262  xid = parsed.twophase_xid;
263 
264  /*
265  * We would like to process the transaction in a two-phase
266  * manner iff output plugin supports two-phase commits and
267  * doesn't filter the transaction at prepare time.
268  */
269  if (info == XLOG_XACT_ABORT_PREPARED)
270  two_phase = !(FilterPrepare(ctx, xid,
271  parsed.twophase_gid));
272 
273  DecodeAbort(ctx, buf, &parsed, xid, two_phase);
274  break;
275  }
277 
278  /*
279  * We assign subxact to the toplevel xact while processing each
280  * record if required. So, we don't need to do anything here. See
281  * LogicalDecodingProcessRecord.
282  */
283  break;
285  {
286  TransactionId xid;
287  xl_xact_invals *invals;
288 
289  xid = XLogRecGetXid(r);
290  invals = (xl_xact_invals *) XLogRecGetData(r);
291 
292  /*
293  * Execute the invalidations for xid-less transactions,
294  * otherwise, accumulate them so that they can be processed at
295  * the commit time.
296  */
297  if (TransactionIdIsValid(xid))
298  {
299  if (!ctx->fast_forward)
300  ReorderBufferAddInvalidations(reorder, xid,
301  buf->origptr,
302  invals->nmsgs,
303  invals->msgs);
305  buf->origptr);
306  }
307  else if ((!ctx->fast_forward))
309  invals->nmsgs,
310  invals->msgs);
311  }
312  break;
313  case XLOG_XACT_PREPARE:
314  {
315  xl_xact_parsed_prepare parsed;
316  xl_xact_prepare *xlrec;
317 
318  /* ok, parse it */
319  xlrec = (xl_xact_prepare *) XLogRecGetData(r);
321  xlrec, &parsed);
322 
323  /*
324  * We would like to process the transaction in a two-phase
325  * manner iff output plugin supports two-phase commits and
326  * doesn't filter the transaction at prepare time.
327  */
328  if (FilterPrepare(ctx, parsed.twophase_xid,
329  parsed.twophase_gid))
330  {
331  ReorderBufferProcessXid(reorder, parsed.twophase_xid,
332  buf->origptr);
333  break;
334  }
335 
336  /*
337  * Note that if the prepared transaction has locked [user]
338  * catalog tables exclusively then decoding prepare can block
339  * till the main transaction is committed because it needs to
340  * lock the catalog tables.
341  *
342  * XXX Now, this can even lead to a deadlock if the prepare
343  * transaction is waiting to get it logically replicated for
344  * distributed 2PC. This can be avoided by disallowing
345  * preparing transactions that have locked [user] catalog
346  * tables exclusively but as of now, we ask users not to do
347  * such an operation.
348  */
349  DecodePrepare(ctx, buf, &parsed);
350  break;
351  }
352  default:
353  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
354  }
355 }
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:840
static bool FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: decode.c:553
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:669
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
Definition: decode.c:765
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:299
int nmsgs
Definition: xact.h:298
TransactionId twophase_xid
Definition: xact.h:421
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:172
#define XLOG_XACT_INVALIDATIONS
Definition: xact.h:175
#define XLOG_XACT_PREPARE
Definition: xact.h:170
#define XLOG_XACT_COMMIT
Definition: xact.h:169
#define XLOG_XACT_OPMASK
Definition: xact.h:179
#define XLOG_XACT_ABORT
Definition: xact.h:171
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:174
#define XLOG_XACT_ABORT_PREPARED
Definition: xact.h:173
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;
156  {
157  xl_parameter_change *xlrec =
159 
160  /*
161  * If wal_level on the primary is reduced to less than
162  * logical, we want to prevent existing logical slots from
163  * being used. Existing logical slots on the standby get
164  * invalidated when this WAL record is replayed; and further,
165  * slot creation fails when wal_level is not sufficient; but
166  * all these operations are not synchronized, so a logical
167  * slot may creep in while the wal_level is being reduced.
168  * Hence this extra check.
169  */
170  if (xlrec->wal_level < WAL_LEVEL_LOGICAL)
171  {
172  /*
173  * This can occur only on a standby, as a primary would
174  * not allow to restart after changing wal_level < logical
175  * if there is pre-existing logical slot.
176  */
178  ereport(ERROR,
179  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
180  errmsg("logical decoding on standby requires wal_level >= logical on the primary")));
181  }
182  break;
183  }
184  case XLOG_NOOP:
185  case XLOG_NEXTOID:
186  case XLOG_SWITCH:
187  case XLOG_BACKUP_END:
188  case XLOG_RESTORE_POINT:
189  case XLOG_FPW_CHANGE:
190  case XLOG_FPI_FOR_HINT:
191  case XLOG_FPI:
194  break;
195  default:
196  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
197  }
198 }
int errcode(int sqlerrcode)
Definition: elog.c:860
int errmsg(const char *fmt,...)
Definition: elog.c:1075
#define ereport(elevel,...)
Definition: elog.h:149
#define XLOG_RESTORE_POINT
Definition: pg_control.h:74
#define XLOG_FPW_CHANGE
Definition: pg_control.h:75
#define XLOG_CHECKPOINT_REDO
Definition: pg_control.h:81
#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:1593
bool RecoveryInProgress(void)
Definition: xlog.c:6211
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:74

References Assert(), buf, elog(), ereport, errcode(), errmsg(), ERROR, RecoveryInProgress(), LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SnapBuildSerializationPoint(), LogicalDecodingContext::snapshot_builder, xl_parameter_change::wal_level, WAL_LEVEL_LOGICAL, XLOG_BACKUP_END, XLOG_CHECKPOINT_ONLINE, XLOG_CHECKPOINT_REDO, 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, XLogRecGetData, XLogRecGetInfo, XLogRecGetXid, and XLR_INFO_MASK.