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 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
uint32 TransactionId
Definition: c.h:652
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
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:768
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:406
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:818
@ 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 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 }
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:573
#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:433
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:708
@ 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 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
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: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.