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

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

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

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

360 {
361  SnapBuild *builder = ctx->snapshot_builder;
362  XLogReaderState *r = buf->record;
363  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
364 
365  ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
366 
367  switch (info)
368  {
369  case XLOG_RUNNING_XACTS:
370  {
372 
373  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
374 
375  /*
376  * Abort all transactions that we keep track of, that are
377  * older than the record's oldestRunningXid. This is the most
378  * convenient spot for doing so since, in contrast to shutdown
379  * or end-of-recovery checkpoints, we have information about
380  * all running transactions which includes prepared ones,
381  * while shutdown checkpoints just know that no non-prepared
382  * transactions are in progress.
383  */
385  }
386  break;
387  case XLOG_STANDBY_LOCK:
388  break;
389  case XLOG_INVALIDATIONS:
390 
391  /*
392  * We are processing the invalidations at the command level via
393  * XLOG_XACT_INVALIDATIONS. So we don't need to do anything here.
394  */
395  break;
396  default:
397  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
398  }
399 }
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1227
#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  }
311  case XLOG_XACT_PREPARE:
312  {
313  xl_xact_parsed_prepare parsed;
314  xl_xact_prepare *xlrec;
315 
316  /* ok, parse it */
317  xlrec = (xl_xact_prepare *) XLogRecGetData(r);
319  xlrec, &parsed);
320 
321  /*
322  * We would like to process the transaction in a two-phase
323  * manner iff output plugin supports two-phase commits and
324  * doesn't filter the transaction at prepare time.
325  */
326  if (FilterPrepare(ctx, parsed.twophase_xid,
327  parsed.twophase_gid))
328  {
329  ReorderBufferProcessXid(reorder, parsed.twophase_xid,
330  buf->origptr);
331  break;
332  }
333 
334  /*
335  * Note that if the prepared transaction has locked [user]
336  * catalog tables exclusively then decoding prepare can block
337  * till the main transaction is committed because it needs to
338  * lock the catalog tables.
339  *
340  * XXX Now, this can even lead to a deadlock if the prepare
341  * transaction is waiting to get it logically replicated for
342  * distributed 2PC. This can be avoided by disallowing
343  * preparing transactions that have locked [user] catalog
344  * tables exclusively but as of now, we ask users not to do
345  * such an operation.
346  */
347  DecodePrepare(ctx, buf, &parsed);
348  break;
349  }
350  default:
351  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
352  }
353 }
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:838
static bool FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: decode.c:551
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:667
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
Definition: decode.c:763
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 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 }
#define Assert(condition)
Definition: c.h:858
int errcode(int sqlerrcode)
Definition: elog.c:853
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define ereport(elevel,...)
Definition: elog.h:149
#define XLOG_RESTORE_POINT
Definition: pg_control.h:75
#define XLOG_FPW_CHANGE
Definition: pg_control.h:76
#define XLOG_CHECKPOINT_REDO
Definition: pg_control.h:82
#define XLOG_OVERWRITE_CONTRECORD
Definition: pg_control.h:81
#define XLOG_FPI
Definition: pg_control.h:79
#define XLOG_FPI_FOR_HINT
Definition: pg_control.h:78
#define XLOG_NEXTOID
Definition: pg_control.h:71
#define XLOG_NOOP
Definition: pg_control.h:70
#define XLOG_CHECKPOINT_SHUTDOWN
Definition: pg_control.h:68
#define XLOG_SWITCH
Definition: pg_control.h:72
#define XLOG_BACKUP_END
Definition: pg_control.h:73
#define XLOG_PARAMETER_CHANGE
Definition: pg_control.h:74
#define XLOG_CHECKPOINT_ONLINE
Definition: pg_control.h:69
#define XLOG_END_OF_RECOVERY
Definition: pg_control.h:77
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1609
bool RecoveryInProgress(void)
Definition: xlog.c:6304
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:76

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.