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

840 {
841  int i;
842  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
843  TimestampTz abort_time = parsed->xact_time;
844  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
845  bool skip_xact;
846 
847  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
848  {
849  origin_lsn = parsed->origin_lsn;
850  abort_time = parsed->origin_timestamp;
851  }
852 
853  /*
854  * Check whether we need to process this transaction. See
855  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
856  * transaction.
857  */
858  skip_xact = DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id);
859 
860  /*
861  * Send the final rollback record for a prepared transaction unless we
862  * need to skip it. For non-two-phase xacts, simply forget the xact.
863  */
864  if (two_phase && !skip_xact)
865  {
866  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
868  abort_time, origin_id, origin_lsn,
869  parsed->twophase_gid, false);
870  }
871  else
872  {
873  for (i = 0; i < parsed->nsubxacts; i++)
874  {
875  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
876  buf->record->EndRecPtr, abort_time);
877  }
878 
879  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr,
880  abort_time);
881  }
882 
883  /* update the decoding stats */
884  UpdateDecodingStats(ctx);
885 }
int64 TimestampTz
Definition: timestamp.h:39
static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid txn_dbid, RepOriginId origin_id)
Definition: decode.c:1299
int i
Definition: isn.c:73
void UpdateDecodingStats(LogicalDecodingContext *ctx)
Definition: logical.c:1931
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 666 of file decode.c.

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

1020 {
1021  XLogReaderState *r = buf->record;
1022  xl_heap_delete *xlrec;
1023  ReorderBufferChange *change;
1024  RelFileLocator target_locator;
1025 
1026  xlrec = (xl_heap_delete *) XLogRecGetData(r);
1027 
1028  /* only interested in our database */
1029  XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
1030  if (target_locator.dbOid != ctx->slot->data.database)
1031  return;
1032 
1033  /* output plugin doesn't look for this origin, no need to queue */
1034  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1035  return;
1036 
1037  change = ReorderBufferGetChange(ctx->reorder);
1038 
1039  if (xlrec->flags & XLH_DELETE_IS_SUPER)
1041  else
1043 
1044  change->origin_id = XLogRecGetOrigin(r);
1045 
1046  memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
1047 
1048  /* old primary key stored */
1049  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
1050  {
1051  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
1052  Size tuplelen = datalen - SizeOfHeapHeader;
1053 
1055 
1056  change->data.tp.oldtuple =
1057  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1058 
1059  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
1060  datalen, change->data.tp.oldtuple);
1061  }
1062 
1063  change->data.tp.clear_toast_afterwards = true;
1064 
1066  change, false);
1067 }
#define Assert(condition)
Definition: c.h:858
size_t Size
Definition: c.h:605
static void DecodeXLogTuple(char *data, Size len, HeapTuple tuple)
Definition: decode.c:1253
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:573
#define XLH_DELETE_CONTAINS_OLD
Definition: heapam_xlog.h:108
#define SizeOfHeapHeader
Definition: heapam_xlog.h:156
#define SizeOfHeapDelete
Definition: heapam_xlog.h:120
#define XLH_DELETE_IS_SUPER
Definition: heapam_xlog.h:104
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:57
@ REORDER_BUFFER_CHANGE_DELETE
Definition: reorderbuffer.h:49
ReplicationSlot * slot
Definition: logical.h:39
ReorderBufferChangeType action
Definition: reorderbuffer.h:76
struct ReorderBufferChange::@105::@106 tp
union ReorderBufferChange::@105 data
RepOriginId origin_id
Definition: reorderbuffer.h:81
ReplicationSlotPersistentData data
Definition: slot.h:178
void XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileLocator *rlocator, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1971
#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 893 of file decode.c.

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

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

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

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

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

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

1074 {
1075  XLogReaderState *r = buf->record;
1076  xl_heap_truncate *xlrec;
1077  ReorderBufferChange *change;
1078 
1079  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
1080 
1081  /* only interested in our database */
1082  if (xlrec->dbId != ctx->slot->data.database)
1083  return;
1084 
1085  /* output plugin doesn't look for this origin, no need to queue */
1086  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1087  return;
1088 
1089  change = ReorderBufferGetChange(ctx->reorder);
1091  change->origin_id = XLogRecGetOrigin(r);
1092  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
1093  change->data.truncate.cascade = true;
1094  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
1095  change->data.truncate.restart_seqs = true;
1096  change->data.truncate.nrelids = xlrec->nrelids;
1097  change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
1098  xlrec->nrelids);
1099  memcpy(change->data.truncate.relids, xlrec->relids,
1100  xlrec->nrelids * sizeof(Oid));
1102  buf->origptr, change, false);
1103 }
#define XLH_TRUNCATE_RESTART_SEQS
Definition: heapam_xlog.h:126
#define XLH_TRUNCATE_CASCADE
Definition: heapam_xlog.h:125
unsigned int Oid
Definition: postgres_ext.h:31
Oid * ReorderBufferGetRelids(ReorderBuffer *rb, int nrelids)
@ REORDER_BUFFER_CHANGE_TRUNCATE
Definition: reorderbuffer.h:58
struct ReorderBufferChange::@105::@107 truncate
Oid relids[FLEXIBLE_ARRAY_MEMBER]
Definition: heapam_xlog.h:138

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

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

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

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

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

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

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

574 {
575  if (ctx->callbacks.filter_by_origin_cb == NULL)
576  return false;
577 
578  return filter_by_origin_cb_wrapper(ctx, origin_id);
579 }
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:1215
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 550 of file decode.c.

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

405 {
406  uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
407  TransactionId xid = XLogRecGetXid(buf->record);
408  SnapBuild *builder = ctx->snapshot_builder;
409 
410  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
411 
412  /*
413  * If we don't have snapshot or we are just fast-forwarding, there is no
414  * point in decoding changes.
415  */
417  ctx->fast_forward)
418  return;
419 
420  switch (info)
421  {
423  if (SnapBuildProcessChange(builder, xid, buf->origptr))
424  DecodeMultiInsert(ctx, buf);
425  break;
426  case XLOG_HEAP2_NEW_CID:
427  {
428  xl_heap_new_cid *xlrec;
429 
430  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
431  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
432 
433  break;
434  }
435  case XLOG_HEAP2_REWRITE:
436 
437  /*
438  * Although these records only exist to serve the needs of logical
439  * decoding, all the work happens as part of crash or archive
440  * recovery, so we don't need to do anything here.
441  */
442  break;
443 
444  /*
445  * Everything else here is just low level physical stuff we're not
446  * interested in.
447  */
451  case XLOG_HEAP2_VISIBLE:
453  break;
454  default:
455  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
456  }
457 }
unsigned char uint8
Definition: c.h:504
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1111
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:224
#define XLOG_HEAP2_MULTI_INSERT
Definition: heapam_xlog.h:63
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:58
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
#define XLOG_HEAP2_PRUNE_VACUUM_SCAN
Definition: heapam_xlog.h:60
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:64
#define XLOG_HEAP2_PRUNE_ON_ACCESS
Definition: heapam_xlog.h:59
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:65
#define XLOG_HEAP2_PRUNE_VACUUM_CLEANUP
Definition: heapam_xlog.h:61
#define XLOG_HEAP2_VISIBLE
Definition: heapam_xlog.h:62
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:768
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:818
@ 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_LOCK_UPDATED, XLOG_HEAP2_MULTI_INSERT, XLOG_HEAP2_NEW_CID, XLOG_HEAP2_PRUNE_ON_ACCESS, XLOG_HEAP2_PRUNE_VACUUM_CLEANUP, XLOG_HEAP2_PRUNE_VACUUM_SCAN, XLOG_HEAP2_REWRITE, XLOG_HEAP2_VISIBLE, XLOG_HEAP_OPMASK, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

◆ heap_decode()

void heap_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 463 of file decode.c.

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

89 {
91  TransactionId txid;
92  RmgrData rmgr;
93 
94  buf.origptr = ctx->reader->ReadRecPtr;
95  buf.endptr = ctx->reader->EndRecPtr;
96  buf.record = record;
97 
98  txid = XLogRecGetTopXid(record);
99 
100  /*
101  * If the top-level xid is valid, we need to assign the subxact to the
102  * top-level xact. We need to do this for all records, hence we do it
103  * before the switch.
104  */
105  if (TransactionIdIsValid(txid))
106  {
108  txid,
109  XLogRecGetXid(record),
110  buf.origptr);
111  }
112 
113  rmgr = GetRmgr(XLogRecGetRmid(record));
114 
115  if (rmgr.rm_decode != NULL)
116  rmgr.rm_decode(ctx, &buf);
117  else
118  {
119  /* just deal with xid, and done */
121  buf.origptr);
122  }
123 }
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(), LogicalSlotAdvanceAndCheckSnapState(), pg_logical_slot_get_changes_guts(), and XLogSendLogical().

◆ logicalmsg_decode()

void logicalmsg_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 585 of file decode.c.

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

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

202 {
203  SnapBuild *builder = ctx->snapshot_builder;
204  ReorderBuffer *reorder = ctx->reorder;
205  XLogReaderState *r = buf->record;
207 
208  /*
209  * If the snapshot isn't yet fully built, we cannot decode anything, so
210  * bail out.
211  */
213  return;
214 
215  switch (info)
216  {
217  case XLOG_XACT_COMMIT:
219  {
220  xl_xact_commit *xlrec;
221  xl_xact_parsed_commit parsed;
222  TransactionId xid;
223  bool two_phase = false;
224 
225  xlrec = (xl_xact_commit *) XLogRecGetData(r);
226  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
227 
228  if (!TransactionIdIsValid(parsed.twophase_xid))
229  xid = XLogRecGetXid(r);
230  else
231  xid = parsed.twophase_xid;
232 
233  /*
234  * We would like to process the transaction in a two-phase
235  * manner iff output plugin supports two-phase commits and
236  * doesn't filter the transaction at prepare time.
237  */
238  if (info == XLOG_XACT_COMMIT_PREPARED)
239  two_phase = !(FilterPrepare(ctx, xid,
240  parsed.twophase_gid));
241 
242  DecodeCommit(ctx, buf, &parsed, xid, two_phase);
243  break;
244  }
245  case XLOG_XACT_ABORT:
247  {
248  xl_xact_abort *xlrec;
249  xl_xact_parsed_abort parsed;
250  TransactionId xid;
251  bool two_phase = false;
252 
253  xlrec = (xl_xact_abort *) XLogRecGetData(r);
254  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
255 
256  if (!TransactionIdIsValid(parsed.twophase_xid))
257  xid = XLogRecGetXid(r);
258  else
259  xid = parsed.twophase_xid;
260 
261  /*
262  * We would like to process the transaction in a two-phase
263  * manner iff output plugin supports two-phase commits and
264  * doesn't filter the transaction at prepare time.
265  */
266  if (info == XLOG_XACT_ABORT_PREPARED)
267  two_phase = !(FilterPrepare(ctx, xid,
268  parsed.twophase_gid));
269 
270  DecodeAbort(ctx, buf, &parsed, xid, two_phase);
271  break;
272  }
274 
275  /*
276  * We assign subxact to the toplevel xact while processing each
277  * record if required. So, we don't need to do anything here. See
278  * LogicalDecodingProcessRecord.
279  */
280  break;
282  {
283  TransactionId xid;
284  xl_xact_invals *invals;
285 
286  xid = XLogRecGetXid(r);
287  invals = (xl_xact_invals *) XLogRecGetData(r);
288 
289  /*
290  * Execute the invalidations for xid-less transactions,
291  * otherwise, accumulate them so that they can be processed at
292  * the commit time.
293  */
294  if (TransactionIdIsValid(xid))
295  {
296  if (!ctx->fast_forward)
297  ReorderBufferAddInvalidations(reorder, xid,
298  buf->origptr,
299  invals->nmsgs,
300  invals->msgs);
302  buf->origptr);
303  }
304  else if ((!ctx->fast_forward))
306  invals->nmsgs,
307  invals->msgs);
308  }
309  break;
310  case XLOG_XACT_PREPARE:
311  {
312  xl_xact_parsed_prepare parsed;
313  xl_xact_prepare *xlrec;
314 
315  /* ok, parse it */
316  xlrec = (xl_xact_prepare *) XLogRecGetData(r);
318  xlrec, &parsed);
319 
320  /*
321  * We would like to process the transaction in a two-phase
322  * manner iff output plugin supports two-phase commits and
323  * doesn't filter the transaction at prepare time.
324  */
325  if (FilterPrepare(ctx, parsed.twophase_xid,
326  parsed.twophase_gid))
327  {
328  ReorderBufferProcessXid(reorder, parsed.twophase_xid,
329  buf->origptr);
330  break;
331  }
332 
333  /*
334  * Note that if the prepared transaction has locked [user]
335  * catalog tables exclusively then decoding prepare can block
336  * till the main transaction is committed because it needs to
337  * lock the catalog tables.
338  *
339  * XXX Now, this can even lead to a deadlock if the prepare
340  * transaction is waiting to get it logically replicated for
341  * distributed 2PC. This can be avoided by disallowing
342  * preparing transactions that have locked [user] catalog
343  * tables exclusively but as of now, we ask users not to do
344  * such an operation.
345  */
346  DecodePrepare(ctx, buf, &parsed);
347  break;
348  }
349  default:
350  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
351  }
352 }
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:837
static bool FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: decode.c:550
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:666
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
Definition: decode.c:762
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 129 of file decode.c.

130 {
131  SnapBuild *builder = ctx->snapshot_builder;
132  uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
133 
135  buf->origptr);
136 
137  switch (info)
138  {
139  /* this is also used in END_OF_RECOVERY checkpoints */
142  SnapBuildSerializationPoint(builder, buf->origptr);
143 
144  break;
146 
147  /*
148  * a RUNNING_XACTS record will have been logged near to this, we
149  * can restart from there.
150  */
151  break;
153  {
154  xl_parameter_change *xlrec =
156 
157  /*
158  * If wal_level on the primary is reduced to less than
159  * logical, we want to prevent existing logical slots from
160  * being used. Existing logical slots on the standby get
161  * invalidated when this WAL record is replayed; and further,
162  * slot creation fails when wal_level is not sufficient; but
163  * all these operations are not synchronized, so a logical
164  * slot may creep in while the wal_level is being reduced.
165  * Hence this extra check.
166  */
167  if (xlrec->wal_level < WAL_LEVEL_LOGICAL)
168  {
169  /*
170  * This can occur only on a standby, as a primary would
171  * not allow to restart after changing wal_level < logical
172  * if there is pre-existing logical slot.
173  */
175  ereport(ERROR,
176  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
177  errmsg("logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary")));
178  }
179  break;
180  }
181  case XLOG_NOOP:
182  case XLOG_NEXTOID:
183  case XLOG_SWITCH:
184  case XLOG_BACKUP_END:
185  case XLOG_RESTORE_POINT:
186  case XLOG_FPW_CHANGE:
187  case XLOG_FPI_FOR_HINT:
188  case XLOG_FPI:
191  break;
192  default:
193  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
194  }
195 }
int errcode(int sqlerrcode)
Definition: elog.c:857
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#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:1592
bool RecoveryInProgress(void)
Definition: xlog.c:6290
@ 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.