PostgreSQL Source Code  git master
decode.h File Reference
Include dependency graph for decode.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  XLogRecordBuffer
 

Typedefs

typedef struct XLogRecordBuffer XLogRecordBuffer
 

Functions

void xlog_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void heap_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void heap2_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void xact_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void standby_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void logicalmsg_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
void LogicalDecodingProcessRecord (LogicalDecodingContext *ctx, XLogReaderState *record)
 

Typedef Documentation

◆ XLogRecordBuffer

Function Documentation

◆ heap2_decode()

void heap2_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 406 of file decode.c.

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

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

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

588 {
589  SnapBuild *builder = ctx->snapshot_builder;
590  XLogReaderState *r = buf->record;
591  TransactionId xid = XLogRecGetXid(r);
592  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
593  RepOriginId origin_id = XLogRecGetOrigin(r);
594  Snapshot snapshot = NULL;
595  xl_logical_message *message;
596 
597  if (info != XLOG_LOGICAL_MESSAGE)
598  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
599 
600  ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
601 
602  /*
603  * If we don't have snapshot or we are just fast-forwarding, there is no
604  * point in decoding messages.
605  */
607  ctx->fast_forward)
608  return;
609 
610  message = (xl_logical_message *) XLogRecGetData(r);
611 
612  if (message->dbId != ctx->slot->data.database ||
613  FilterByOrigin(ctx, origin_id))
614  return;
615 
616  if (message->transactional &&
617  !SnapBuildProcessChange(builder, xid, buf->origptr))
618  return;
619  else if (!message->transactional &&
621  SnapBuildXactNeedsSkip(builder, buf->origptr)))
622  return;
623 
624  /*
625  * If this is a non-transactional change, get the snapshot we're expected
626  * to use. We only get here when the snapshot is consistent, and the
627  * change is not meant to be skipped.
628  *
629  * For transactional changes we don't need a snapshot, we'll use the
630  * regular snapshot maintained by ReorderBuffer. We just leave it NULL.
631  */
632  if (!message->transactional)
633  snapshot = SnapBuildGetOrBuildSnapshot(builder);
634 
635  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
636  message->transactional,
637  message->message, /* first part of message is
638  * prefix */
639  message->message_size,
640  message->message + message->prefix_size);
641 }
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:575
#define XLOG_LOGICAL_MESSAGE
Definition: message.h:36
void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snap, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:434
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:709
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:46
ReplicationSlot * slot
Definition: logical.h:39
ReplicationSlotPersistentData data
Definition: slot.h:162
bool transactional
Definition: message.h:23
char message[FLEXIBLE_ARRAY_MEMBER]
Definition: message.h:27
uint16 RepOriginId
Definition: xlogdefs.h:65
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:413
#define XLR_INFO_MASK
Definition: xlogrecord.h:62

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

◆ standby_decode()

void standby_decode ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)

Definition at line 360 of file decode.c.

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

204 {
205  SnapBuild *builder = ctx->snapshot_builder;
206  ReorderBuffer *reorder = ctx->reorder;
207  XLogReaderState *r = buf->record;
209 
210  /*
211  * If the snapshot isn't yet fully built, we cannot decode anything, so
212  * bail out.
213  */
215  return;
216 
217  switch (info)
218  {
219  case XLOG_XACT_COMMIT:
221  {
222  xl_xact_commit *xlrec;
223  xl_xact_parsed_commit parsed;
224  TransactionId xid;
225  bool two_phase = false;
226 
227  xlrec = (xl_xact_commit *) XLogRecGetData(r);
228  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
229 
230  if (!TransactionIdIsValid(parsed.twophase_xid))
231  xid = XLogRecGetXid(r);
232  else
233  xid = parsed.twophase_xid;
234 
235  /*
236  * We would like to process the transaction in a two-phase
237  * manner iff output plugin supports two-phase commits and
238  * doesn't filter the transaction at prepare time.
239  */
240  if (info == XLOG_XACT_COMMIT_PREPARED)
241  two_phase = !(FilterPrepare(ctx, xid,
242  parsed.twophase_gid));
243 
244  DecodeCommit(ctx, buf, &parsed, xid, two_phase);
245  break;
246  }
247  case XLOG_XACT_ABORT:
249  {
250  xl_xact_abort *xlrec;
251  xl_xact_parsed_abort parsed;
252  TransactionId xid;
253  bool two_phase = false;
254 
255  xlrec = (xl_xact_abort *) XLogRecGetData(r);
256  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
257 
258  if (!TransactionIdIsValid(parsed.twophase_xid))
259  xid = XLogRecGetXid(r);
260  else
261  xid = parsed.twophase_xid;
262 
263  /*
264  * We would like to process the transaction in a two-phase
265  * manner iff output plugin supports two-phase commits and
266  * doesn't filter the transaction at prepare time.
267  */
268  if (info == XLOG_XACT_ABORT_PREPARED)
269  two_phase = !(FilterPrepare(ctx, xid,
270  parsed.twophase_gid));
271 
272  DecodeAbort(ctx, buf, &parsed, xid, two_phase);
273  break;
274  }
276 
277  /*
278  * We assign subxact to the toplevel xact while processing each
279  * record if required. So, we don't need to do anything here. See
280  * LogicalDecodingProcessRecord.
281  */
282  break;
284  {
285  TransactionId xid;
286  xl_xact_invals *invals;
287 
288  xid = XLogRecGetXid(r);
289  invals = (xl_xact_invals *) XLogRecGetData(r);
290 
291  /*
292  * Execute the invalidations for xid-less transactions,
293  * otherwise, accumulate them so that they can be processed at
294  * the commit time.
295  */
296  if (TransactionIdIsValid(xid))
297  {
298  if (!ctx->fast_forward)
299  ReorderBufferAddInvalidations(reorder, xid,
300  buf->origptr,
301  invals->nmsgs,
302  invals->msgs);
304  buf->origptr);
305  }
306  else if ((!ctx->fast_forward))
308  invals->nmsgs,
309  invals->msgs);
310  }
311  break;
312  case XLOG_XACT_PREPARE:
313  {
314  xl_xact_parsed_prepare parsed;
315  xl_xact_prepare *xlrec;
316 
317  /* ok, parse it */
318  xlrec = (xl_xact_prepare *) XLogRecGetData(r);
320  xlrec, &parsed);
321 
322  /*
323  * We would like to process the transaction in a two-phase
324  * manner iff output plugin supports two-phase commits and
325  * doesn't filter the transaction at prepare time.
326  */
327  if (FilterPrepare(ctx, parsed.twophase_xid,
328  parsed.twophase_gid))
329  {
330  ReorderBufferProcessXid(reorder, parsed.twophase_xid,
331  buf->origptr);
332  break;
333  }
334 
335  /*
336  * Note that if the prepared transaction has locked [user]
337  * catalog tables exclusively then decoding prepare can block
338  * till the main transaction is committed because it needs to
339  * lock the catalog tables.
340  *
341  * XXX Now, this can even lead to a deadlock if the prepare
342  * transaction is waiting to get it logically replicated for
343  * distributed 2PC. This can be avoided by disallowing
344  * preparing transactions that have locked [user] catalog
345  * tables exclusively but as of now, we ask users not to do
346  * such an operation.
347  */
348  DecodePrepare(ctx, buf, &parsed);
349  break;
350  }
351  default:
352  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
353  }
354 }
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:823
static bool FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: decode.c:552
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:652
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
Definition: decode.c:748
static bool two_phase
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
char twophase_gid[GIDSIZE]
Definition: xact.h:422
TransactionId twophase_xid
Definition: xact.h:391
char twophase_gid[GIDSIZE]
Definition: xact.h:392
#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:
193  break;
194  default:
195  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
196  }
197 }
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define ereport(elevel,...)
Definition: elog.h:149
Assert(fmt[strlen(fmt) - 1] !='\n')
#define XLOG_RESTORE_POINT
Definition: pg_control.h:74
#define XLOG_FPW_CHANGE
Definition: pg_control.h:75
#define XLOG_OVERWRITE_CONTRECORD
Definition: pg_control.h:80
#define XLOG_FPI
Definition: pg_control.h:78
#define XLOG_FPI_FOR_HINT
Definition: pg_control.h:77
#define XLOG_NEXTOID
Definition: pg_control.h:70
#define XLOG_NOOP
Definition: pg_control.h:69
#define XLOG_CHECKPOINT_SHUTDOWN
Definition: pg_control.h:67
#define XLOG_SWITCH
Definition: pg_control.h:71
#define XLOG_BACKUP_END
Definition: pg_control.h:72
#define XLOG_PARAMETER_CHANGE
Definition: pg_control.h:73
#define XLOG_CHECKPOINT_ONLINE
Definition: pg_control.h:68
#define XLOG_END_OF_RECOVERY
Definition: pg_control.h:76
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1593
bool RecoveryInProgress(void)
Definition: xlog.c:5948
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:71

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_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.