PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
decode.c File Reference
Include dependency graph for decode.c:

Go to the source code of this file.

Data Structures

struct  XLogRecordBuffer
 

Typedefs

typedef struct XLogRecordBuffer XLogRecordBuffer
 

Functions

static void DecodeXLogOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeHeapOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeHeap2Op (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeXactOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeStandbyOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeLogicalMsgOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeInsert (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeUpdate (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
static void DecodeDelete (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)
 
static void DecodeAbort (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid)
 
static void DecodeXLogTuple (char *data, Size len, ReorderBufferTupleBuf *tup)
 
void LogicalDecodingProcessRecord (LogicalDecodingContext *ctx, XLogReaderState *record)
 
static bool FilterByOrigin (LogicalDecodingContext *ctx, RepOriginId origin_id)
 

Typedef Documentation

Function Documentation

static void DecodeAbort ( LogicalDecodingContext ctx,
XLogRecordBuffer buf,
xl_xact_parsed_abort parsed,
TransactionId  xid 
)
static

Definition at line 620 of file decode.c.

References XLogReaderState::EndRecPtr, i, xl_xact_parsed_abort::nsubxacts, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAbort(), and xl_xact_parsed_abort::subxacts.

Referenced by DecodeXactOp().

622 {
623  int i;
624 
625  for (i = 0; i < parsed->nsubxacts; i++)
626  {
627  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
628  buf->record->EndRecPtr);
629  }
630 
631  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
632 }
struct ReorderBuffer * reorder
Definition: logical.h:45
XLogRecPtr EndRecPtr
Definition: xlogreader.h:115
XLogReaderState * record
Definition: decode.c:53
TransactionId * subxacts
Definition: xact.h:317
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
int i
static void DecodeCommit ( LogicalDecodingContext ctx,
XLogRecordBuffer buf,
xl_xact_parsed_commit parsed,
TransactionId  xid 
)
static

Definition at line 536 of file decode.c.

References ReplicationSlot::data, ReplicationSlotPersistentData::database, xl_xact_parsed_commit::dbId, XLogRecordBuffer::endptr, FilterByOrigin(), i, InvalidOid, InvalidXLogRecPtr, xl_xact_parsed_commit::msgs, xl_xact_parsed_commit::nmsgs, xl_xact_parsed_commit::nsubxacts, xl_xact_parsed_commit::origin_lsn, xl_xact_parsed_commit::origin_timestamp, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAddInvalidations(), ReorderBufferCommit(), ReorderBufferCommitChild(), ReorderBufferForget(), ReorderBufferXidSetCatalogChanges(), LogicalDecodingContext::slot, SnapBuildCommitTxn(), SnapBuildXactNeedsSkip(), LogicalDecodingContext::snapshot_builder, xl_xact_parsed_commit::subxacts, xl_xact_parsed_commit::xact_time, XACT_XINFO_HAS_ORIGIN, xl_xact_parsed_commit::xinfo, and XLogRecGetOrigin.

Referenced by DecodeXactOp().

538 {
539  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
540  TimestampTz commit_time = parsed->xact_time;
541  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
542  int i;
543 
544  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
545  {
546  origin_lsn = parsed->origin_lsn;
547  commit_time = parsed->origin_timestamp;
548  }
549 
550  /*
551  * Process invalidation messages, even if we're not interested in the
552  * transaction's contents, since the various caches need to always be
553  * consistent.
554  */
555  if (parsed->nmsgs > 0)
556  {
558  parsed->nmsgs, parsed->msgs);
560  }
561 
563  parsed->nsubxacts, parsed->subxacts);
564 
565  /* ----
566  * Check whether we are interested in this specific transaction, and tell
567  * the reorderbuffer to forget the content of the (sub-)transactions
568  * if not.
569  *
570  * There can be several reasons we might not be interested in this
571  * transaction:
572  * 1) We might not be interested in decoding transactions up to this
573  * LSN. This can happen because we previously decoded it and now just
574  * are restarting or if we haven't assembled a consistent snapshot yet.
575  * 2) The transaction happened in another database.
576  * 3) The output plugin is not interested in the origin.
577  *
578  * We can't just use ReorderBufferAbort() here, because we need to execute
579  * the transaction's invalidations. This currently won't be needed if
580  * we're just skipping over the transaction because currently we only do
581  * so during startup, to get to the first transaction the client needs. As
582  * we have reset the catalog caches before starting to read WAL, and we
583  * haven't yet touched any catalogs, there can't be anything to invalidate.
584  * But if we're "forgetting" this commit because it's it happened in
585  * another database, the invalidations might be important, because they
586  * could be for shared catalogs and we might have loaded data into the
587  * relevant syscaches.
588  * ---
589  */
591  (parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
592  FilterByOrigin(ctx, origin_id))
593  {
594  for (i = 0; i < parsed->nsubxacts; i++)
595  {
596  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
597  }
598  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
599 
600  return;
601  }
602 
603  /* tell the reorderbuffer about the surviving subtransactions */
604  for (i = 0; i < parsed->nsubxacts; i++)
605  {
606  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
607  buf->origptr, buf->endptr);
608  }
609 
610  /* replay actions of all transaction + subtransactions in order */
611  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
612  commit_time, origin_id, origin_lsn);
613 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:158
struct ReorderBuffer * reorder
Definition: logical.h:45
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:410
void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
TimestampTz origin_timestamp
Definition: xact.h:308
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:927
int64 TimestampTz
Definition: timestamp.h:39
TransactionId * subxacts
Definition: xact.h:297
uint16 RepOriginId
Definition: xlogdefs.h:51
SharedInvalidationMessage * msgs
Definition: xact.h:303
XLogRecPtr origin_lsn
Definition: xact.h:307
ReplicationSlotPersistentData data
Definition: slot.h:115
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:477
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:219
XLogRecPtr endptr
Definition: decode.c:52
XLogReaderState * record
Definition: decode.c:53
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
#define InvalidOid
Definition: postgres_ext.h:36
struct SnapBuild * snapshot_builder
Definition: logical.h:46
uint64 XLogRecPtr
Definition: xlogdefs.h:21
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
XLogRecPtr origptr
Definition: decode.c:51
ReplicationSlot * slot
Definition: logical.h:41
int i
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
TimestampTz xact_time
Definition: xact.h:289
static void DecodeDelete ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 757 of file decode.c.

References ReorderBufferChange::action, Assert, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_delete::flags, NULL, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_DELETE, 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 DecodeHeapOp().

758 {
759  XLogReaderState *r = buf->record;
760  xl_heap_delete *xlrec;
761  ReorderBufferChange *change;
762  RelFileNode target_node;
763 
764  xlrec = (xl_heap_delete *) XLogRecGetData(r);
765 
766  /* only interested in our database */
767  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
768  if (target_node.dbNode != ctx->slot->data.database)
769  return;
770 
771  /*
772  * Super deletions are irrelevant for logical decoding, it's driven by the
773  * confirmation records.
774  */
775  if (xlrec->flags & XLH_DELETE_IS_SUPER)
776  return;
777 
778  /* output plugin doesn't look for this origin, no need to queue */
779  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
780  return;
781 
782  change = ReorderBufferGetChange(ctx->reorder);
784  change->origin_id = XLogRecGetOrigin(r);
785 
786  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
787 
788  /* old primary key stored */
789  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
790  {
791  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
792  Size tuplelen = datalen - SizeOfHeapHeader;
793 
794  Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
795 
796  change->data.tp.oldtuple =
797  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
798 
799  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
800  datalen, change->data.tp.oldtuple);
801  }
802 
803  change->data.tp.clear_toast_afterwards = true;
804 
805  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
806 }
#define XLH_DELETE_IS_SUPER
Definition: heapam_xlog.h:95
struct ReorderBuffer * reorder
Definition: logical.h:45
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:954
union ReorderBufferChange::@93 data
#define XLH_DELETE_CONTAINS_OLD
Definition: heapam_xlog.h:98
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:115
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:477
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:77
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:219
#define XLogRecGetData(decoder)
Definition: xlogreader.h:220
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:221
struct ReorderBufferChange::@93::@94 tp
#define SizeOfHeapDelete
Definition: heapam_xlog.h:110
XLogReaderState * record
Definition: decode.c:53
RepOriginId origin_id
Definition: reorderbuffer.h:79
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1307
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
size_t Size
Definition: c.h:356
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:51
ReplicationSlot * slot
Definition: logical.h:41
#define SizeOfHeapHeader
Definition: heapam_xlog.h:127
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
static void DecodeHeap2Op ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 348 of file decode.c.

References DecodeMultiInsert(), elog, ERROR, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), SnapBuildProcessChange(), SnapBuildProcessNewCid(), LogicalDecodingContext::snapshot_builder, XLOG_HEAP2_CLEAN, XLOG_HEAP2_CLEANUP_INFO, XLOG_HEAP2_FREEZE_PAGE, XLOG_HEAP2_LOCK_UPDATED, XLOG_HEAP2_MULTI_INSERT, XLOG_HEAP2_NEW_CID, XLOG_HEAP2_REWRITE, XLOG_HEAP2_VISIBLE, XLOG_HEAP_OPMASK, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

Referenced by LogicalDecodingProcessRecord().

349 {
351  TransactionId xid = XLogRecGetXid(buf->record);
352  SnapBuild *builder = ctx->snapshot_builder;
353 
354  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
355 
356  /* no point in doing anything yet */
358  return;
359 
360  switch (info)
361  {
363  if (SnapBuildProcessChange(builder, xid, buf->origptr))
364  DecodeMultiInsert(ctx, buf);
365  break;
366  case XLOG_HEAP2_NEW_CID:
367  {
368  xl_heap_new_cid *xlrec;
369 
370  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
371  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
372 
373  break;
374  }
375  case XLOG_HEAP2_REWRITE:
376 
377  /*
378  * Although these records only exist to serve the needs of logical
379  * decoding, all the work happens as part of crash or archive
380  * recovery, so we don't need to do anything here.
381  */
382  break;
383 
384  /*
385  * Everything else here is just low level physical stuff we're not
386  * interested in.
387  */
389  case XLOG_HEAP2_CLEAN:
391  case XLOG_HEAP2_VISIBLE:
393  break;
394  default:
395  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
396  }
397 }
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:814
struct ReorderBuffer * reorder
Definition: logical.h:45
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:59
uint32 TransactionId
Definition: c.h:397
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:401
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:53
unsigned char uint8
Definition: c.h:266
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
#define XLOG_HEAP2_MULTI_INSERT
Definition: heapam_xlog.h:58
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:719
#define XLOG_HEAP2_VISIBLE
Definition: heapam_xlog.h:57
#define XLogRecGetData(decoder)
Definition: xlogreader.h:220
#define ERROR
Definition: elog.h:43
#define XLOG_HEAP2_CLEAN
Definition: heapam_xlog.h:54
#define XLOG_HEAP2_CLEANUP_INFO
Definition: heapam_xlog.h:56
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:769
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:60
XLogReaderState * record
Definition: decode.c:53
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:216
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
struct SnapBuild * snapshot_builder
Definition: logical.h:46
#define XLOG_HEAP2_FREEZE_PAGE
Definition: heapam_xlog.h:55
XLogRecPtr origptr
Definition: decode.c:51
#define elog
Definition: elog.h:219
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
static void DecodeHeapOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 403 of file decode.c.

References DecodeDelete(), DecodeInsert(), DecodeSpecConfirm(), DecodeUpdate(), elog, ERROR, XLogRecordBuffer::origptr, XLogRecordBuffer::record, 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_UPDATE, XLogRecGetInfo, and XLogRecGetXid.

Referenced by LogicalDecodingProcessRecord().

404 {
406  TransactionId xid = XLogRecGetXid(buf->record);
407  SnapBuild *builder = ctx->snapshot_builder;
408 
409  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
410 
411  /* no point in doing anything yet */
413  return;
414 
415  switch (info)
416  {
417  case XLOG_HEAP_INSERT:
418  if (SnapBuildProcessChange(builder, xid, buf->origptr))
419  DecodeInsert(ctx, buf);
420  break;
421 
422  /*
423  * Treat HOT update as normal updates. There is no useful
424  * information in the fact that we could make it a HOT update
425  * locally and the WAL layout is compatible.
426  */
428  case XLOG_HEAP_UPDATE:
429  if (SnapBuildProcessChange(builder, xid, buf->origptr))
430  DecodeUpdate(ctx, buf);
431  break;
432 
433  case XLOG_HEAP_DELETE:
434  if (SnapBuildProcessChange(builder, xid, buf->origptr))
435  DecodeDelete(ctx, buf);
436  break;
437 
438  case XLOG_HEAP_INPLACE:
439 
440  /*
441  * Inplace updates are only ever performed on catalog tuples and
442  * can, per definition, not change tuple visibility. Since we
443  * don't decode catalog tuples, we're not interested in the
444  * record's contents.
445  *
446  * In-place updates can be used either by XID-bearing transactions
447  * (e.g. in CREATE INDEX CONCURRENTLY) or by XID-less
448  * transactions (e.g. VACUUM). In the former case, the commit
449  * record will include cache invalidations, so we mark the
450  * transaction as catalog modifying here. Currently that's
451  * redundant because the commit will do that as well, but once we
452  * support decoding in-progress relations, this will be important.
453  */
454  if (!TransactionIdIsValid(xid))
455  break;
456 
457  SnapBuildProcessChange(builder, xid, buf->origptr);
459  break;
460 
461  case XLOG_HEAP_CONFIRM:
462  if (SnapBuildProcessChange(builder, xid, buf->origptr))
463  DecodeSpecConfirm(ctx, buf);
464  break;
465 
466  case XLOG_HEAP_LOCK:
467  /* we don't care about row level locks for now */
468  break;
469 
470  default:
471  elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
472  break;
473  }
474 }
#define XLOG_HEAP_INSERT
Definition: heapam_xlog.h:32
struct ReorderBuffer * reorder
Definition: logical.h:45
static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:640
uint32 TransactionId
Definition: c.h:397
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:401
unsigned char uint8
Definition: c.h:266
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
#define XLOG_HEAP_HOT_UPDATE
Definition: heapam_xlog.h:36
static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:691
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:719
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:919
#define ERROR
Definition: elog.h:43
static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:757
XLogReaderState * record
Definition: decode.c:53
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:216
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
struct SnapBuild * snapshot_builder
Definition: logical.h:46
#define XLOG_HEAP_UPDATE
Definition: heapam_xlog.h:34
#define XLOG_HEAP_INPLACE
Definition: heapam_xlog.h:39
XLogRecPtr origptr
Definition: decode.c:51
#define XLOG_HEAP_LOCK
Definition: heapam_xlog.h:38
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define elog
Definition: elog.h:219
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define XLOG_HEAP_DELETE
Definition: heapam_xlog.h:33
#define XLOG_HEAP_CONFIRM
Definition: heapam_xlog.h:37
static void DecodeInsert ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 640 of file decode.c.

References ReorderBufferChange::action, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_insert::flags, NULL, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, 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, XLogRecGetBlockData(), XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeapOp().

641 {
642  XLogReaderState *r = buf->record;
643  xl_heap_insert *xlrec;
644  ReorderBufferChange *change;
645  RelFileNode target_node;
646 
647  xlrec = (xl_heap_insert *) XLogRecGetData(r);
648 
649  /* only interested in our database */
650  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
651  if (target_node.dbNode != ctx->slot->data.database)
652  return;
653 
654  /* output plugin doesn't look for this origin, no need to queue */
655  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
656  return;
657 
658  change = ReorderBufferGetChange(ctx->reorder);
659  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
661  else
663  change->origin_id = XLogRecGetOrigin(r);
664 
665  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
666 
668  {
669  Size datalen;
670  char *tupledata = XLogRecGetBlockData(r, 0, &datalen);
671  Size tuplelen = datalen - SizeOfHeapHeader;
672 
673  change->data.tp.newtuple =
674  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
675 
676  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
677  }
678 
679  change->data.tp.clear_toast_afterwards = true;
680 
681  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
682 }
struct ReorderBuffer * reorder
Definition: logical.h:45
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:954
union ReorderBufferChange::@93 data
#define XLH_INSERT_IS_SPECULATIVE
Definition: heapam_xlog.h:68
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:115
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:477
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:77
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:219
#define XLogRecGetData(decoder)
Definition: xlogreader.h:220
#define XLH_INSERT_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:69
struct ReorderBufferChange::@93::@94 tp
XLogReaderState * record
Definition: decode.c:53
RepOriginId origin_id
Definition: reorderbuffer.h:79
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1307
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1331
#define NULL
Definition: c.h:229
size_t Size
Definition: c.h:356
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:51
ReplicationSlot * slot
Definition: logical.h:41
#define SizeOfHeapHeader
Definition: heapam_xlog.h:127
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
static void DecodeLogicalMsgOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 489 of file decode.c.

References ReplicationSlot::data, ReplicationSlotPersistentData::database, xl_logical_message::dbId, elog, XLogRecordBuffer::endptr, ERROR, FilterByOrigin(), xl_logical_message::message, xl_logical_message::message_size, XLogRecordBuffer::origptr, xl_logical_message::prefix_size, XLogRecordBuffer::record, 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.

Referenced by LogicalDecodingProcessRecord().

490 {
491  SnapBuild *builder = ctx->snapshot_builder;
492  XLogReaderState *r = buf->record;
493  TransactionId xid = XLogRecGetXid(r);
494  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
495  RepOriginId origin_id = XLogRecGetOrigin(r);
496  Snapshot snapshot;
497  xl_logical_message *message;
498 
499  if (info != XLOG_LOGICAL_MESSAGE)
500  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
501 
503 
504  /* No point in doing anything yet. */
506  return;
507 
508  message = (xl_logical_message *) XLogRecGetData(r);
509 
510  if (message->dbId != ctx->slot->data.database ||
511  FilterByOrigin(ctx, origin_id))
512  return;
513 
514  if (message->transactional &&
515  !SnapBuildProcessChange(builder, xid, buf->origptr))
516  return;
517  else if (!message->transactional &&
519  SnapBuildXactNeedsSkip(builder, buf->origptr)))
520  return;
521 
522  snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
523  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
524  message->transactional,
525  message->message, /* first part of message is
526  * prefix */
527  message->message_size,
528  message->message + message->prefix_size);
529 }
void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snapshot, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
struct ReorderBuffer * reorder
Definition: logical.h:45
uint32 TransactionId
Definition: c.h:397
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:410
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:401
unsigned char uint8
Definition: c.h:266
uint16 RepOriginId
Definition: xlogdefs.h:51
ReplicationSlotPersistentData data
Definition: slot.h:115
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:477
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:219
char message[FLEXIBLE_ARRAY_MEMBER]
Definition: message.h:26
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:719
#define XLogRecGetData(decoder)
Definition: xlogreader.h:220
#define ERROR
Definition: elog.h:43
XLogRecPtr endptr
Definition: decode.c:52
XLogReaderState * record
Definition: decode.c:53
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:216
bool transactional
Definition: message.h:23
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
struct SnapBuild * snapshot_builder
Definition: logical.h:46
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
XLogRecPtr origptr
Definition: decode.c:51
ReplicationSlot * slot
Definition: logical.h:41
#define XLOG_LOGICAL_MESSAGE
Definition: message.h:37
#define elog
Definition: elog.h:219
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:675
static void DecodeMultiInsert ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 814 of file decode.c.

References ReorderBufferChange::action, Assert, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, xl_multi_insert_tuple::datalen, RelFileNode::dbNode, FilterByOrigin(), xl_heap_multi_insert::flags, header(), i, InvalidOid, ItemPointerSetInvalid, xl_heap_multi_insert::ntuples, NULL, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INSERT, ReorderBufferGetChange(), ReorderBufferGetTupleBuf(), ReorderBufferQueueChange(), SHORTALIGN, SizeofHeapTupleHeader, SizeOfMultiInsertTuple, LogicalDecodingContext::slot, HeapTupleData::t_data, HeapTupleHeaderData::t_hoff, xl_multi_insert_tuple::t_hoff, HeapTupleHeaderData::t_infomask, xl_multi_insert_tuple::t_infomask, HeapTupleHeaderData::t_infomask2, xl_multi_insert_tuple::t_infomask2, HeapTupleData::t_len, HeapTupleData::t_self, HeapTupleData::t_tableOid, ReorderBufferChange::tp, ReorderBufferTupleBuf::tuple, XLH_INSERT_CONTAINS_NEW_TUPLE, XLH_INSERT_LAST_IN_MULTI, XLogRecGetBlockData(), XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeap2Op().

815 {
816  XLogReaderState *r = buf->record;
817  xl_heap_multi_insert *xlrec;
818  int i;
819  char *data;
820  char *tupledata;
821  Size tuplelen;
822  RelFileNode rnode;
823 
824  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
825 
826  /* only interested in our database */
827  XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
828  if (rnode.dbNode != ctx->slot->data.database)
829  return;
830 
831  /* output plugin doesn't look for this origin, no need to queue */
832  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
833  return;
834 
835  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
836 
837  data = tupledata;
838  for (i = 0; i < xlrec->ntuples; i++)
839  {
840  ReorderBufferChange *change;
841  xl_multi_insert_tuple *xlhdr;
842  int datalen;
843  ReorderBufferTupleBuf *tuple;
844 
845  change = ReorderBufferGetChange(ctx->reorder);
847  change->origin_id = XLogRecGetOrigin(r);
848 
849  memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
850 
851  /*
852  * CONTAINS_NEW_TUPLE will always be set currently as multi_insert
853  * isn't used for catalogs, but better be future proof.
854  *
855  * We decode the tuple in pretty much the same way as DecodeXLogTuple,
856  * but since the layout is slightly different, we can't use it here.
857  */
859  {
861 
862  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
863  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
864  datalen = xlhdr->datalen;
865 
866  change->data.tp.newtuple =
867  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
868 
869  tuple = change->data.tp.newtuple;
870  header = tuple->tuple.t_data;
871 
872  /* not a disk based tuple */
874 
875  /*
876  * We can only figure this out after reassembling the
877  * transactions.
878  */
879  tuple->tuple.t_tableOid = InvalidOid;
880 
881  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
882 
883  memset(header, 0, SizeofHeapTupleHeader);
884 
885  memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
886  (char *) data,
887  datalen);
888  data += datalen;
889 
890  header->t_infomask = xlhdr->t_infomask;
891  header->t_infomask2 = xlhdr->t_infomask2;
892  header->t_hoff = xlhdr->t_hoff;
893  }
894 
895  /*
896  * Reset toast reassembly state only after the last row in the last
897  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
898  * call.
899  */
900  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
901  (i + 1) == xlrec->ntuples)
902  change->data.tp.clear_toast_afterwards = true;
903  else
904  change->data.tp.clear_toast_afterwards = false;
905 
907  buf->origptr, change);
908  }
909  Assert(data == tupledata + tuplelen);
910 }
#define SizeofHeapTupleHeader
Definition: htup_details.h:170
struct ReorderBuffer * reorder
Definition: logical.h:45
union ReorderBufferChange::@93 data
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:115
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:477
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:77
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:219
HeapTupleHeader t_data
Definition: htup.h:67
#define XLogRecGetData(decoder)
Definition: xlogreader.h:220
#define XLH_INSERT_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:69
ItemPointerData t_self
Definition: htup.h:65
struct ReorderBufferChange::@93::@94 tp
uint32 t_len
Definition: htup.h:64
HeapTupleData tuple
Definition: reorderbuffer.h:27
Oid t_tableOid
Definition: htup.h:66
XLogReaderState * record
Definition: decode.c:53
RepOriginId origin_id
Definition: reorderbuffer.h:79
#define XLH_INSERT_LAST_IN_MULTI
Definition: heapam_xlog.h:67
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1307
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1331
#define InvalidOid
Definition: postgres_ext.h:36
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
size_t Size
Definition: c.h:356
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:207
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:51
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:149
ReplicationSlot * slot
Definition: logical.h:41
int i
#define SHORTALIGN(LEN)
Definition: c.h:584
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:169
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
static void DecodeSpecConfirm ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 919 of file decode.c.

References ReorderBufferChange::action, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, FilterByOrigin(), NULL, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, ReorderBufferGetChange(), ReorderBufferQueueChange(), LogicalDecodingContext::slot, ReorderBufferChange::tp, XLogRecGetBlockTag(), XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeapOp().

920 {
921  XLogReaderState *r = buf->record;
922  ReorderBufferChange *change;
923  RelFileNode target_node;
924 
925  /* only interested in our database */
926  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
927  if (target_node.dbNode != ctx->slot->data.database)
928  return;
929 
930  /* output plugin doesn't look for this origin, no need to queue */
931  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
932  return;
933 
934  change = ReorderBufferGetChange(ctx->reorder);
936  change->origin_id = XLogRecGetOrigin(r);
937 
938  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
939 
940  change->data.tp.clear_toast_afterwards = true;
941 
942  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
943 }
struct ReorderBuffer * reorder
Definition: logical.h:45
union ReorderBufferChange::@93 data
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:115
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:477
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:77
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:219
struct ReorderBufferChange::@93::@94 tp
XLogReaderState * record
Definition: decode.c:53
RepOriginId origin_id
Definition: reorderbuffer.h:79
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1307
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
#define NULL
Definition: c.h:229
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:51
ReplicationSlot * slot
Definition: logical.h:41
static void DecodeStandbyOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 300 of file decode.c.

References elog, ERROR, xl_invalidations::msgs, xl_invalidations::nmsgs, xl_running_xacts::oldestRunningXid, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAbortOld(), ReorderBufferImmediateInvalidation(), ReorderBufferProcessXid(), SnapBuildProcessRunningXacts(), LogicalDecodingContext::snapshot_builder, XLOG_INVALIDATIONS, XLOG_RUNNING_XACTS, XLOG_STANDBY_LOCK, XLogRecGetData, XLogRecGetInfo, XLogRecGetXid, and XLR_INFO_MASK.

Referenced by LogicalDecodingProcessRecord().

301 {
302  SnapBuild *builder = ctx->snapshot_builder;
303  XLogReaderState *r = buf->record;
304  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
305 
307 
308  switch (info)
309  {
310  case XLOG_RUNNING_XACTS:
311  {
313 
314  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
315 
316  /*
317  * Abort all transactions that we keep track of, that are
318  * older than the record's oldestRunningXid. This is the most
319  * convenient spot for doing so since, in contrast to shutdown
320  * or end-of-recovery checkpoints, we have information about
321  * all running transactions which includes prepared ones,
322  * while shutdown checkpoints just know that no non-prepared
323  * transactions are in progress.
324  */
326  }
327  break;
328  case XLOG_STANDBY_LOCK:
329  break;
330  case XLOG_INVALIDATIONS:
331  {
332  xl_invalidations *invalidations =
334 
336  ctx->reorder, invalidations->nmsgs, invalidations->msgs);
337  }
338  break;
339  default:
340  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
341  }
342 }
struct ReorderBuffer * reorder
Definition: logical.h:45
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
#define XLOG_INVALIDATIONS
Definition: standbydefs.h:36
unsigned char uint8
Definition: c.h:266
#define XLOG_STANDBY_LOCK
Definition: standbydefs.h:34
#define XLogRecGetData(decoder)
Definition: xlogreader.h:220
#define ERROR
Definition: elog.h:43
void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1094
XLogReaderState * record
Definition: decode.c:53
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:216
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
struct SnapBuild * snapshot_builder
Definition: logical.h:46
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER]
Definition: standbydefs.h:69
XLogRecPtr origptr
Definition: decode.c:51
TransactionId oldestRunningXid
Definition: standbydefs.h:53
#define XLOG_RUNNING_XACTS
Definition: standbydefs.h:35
#define elog
Definition: elog.h:219
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
static void DecodeUpdate ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 691 of file decode.c.

References ReorderBufferChange::action, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_update::flags, NULL, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, 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 DecodeHeapOp().

692 {
693  XLogReaderState *r = buf->record;
694  xl_heap_update *xlrec;
695  ReorderBufferChange *change;
696  char *data;
697  RelFileNode target_node;
698 
699  xlrec = (xl_heap_update *) XLogRecGetData(r);
700 
701  /* only interested in our database */
702  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
703  if (target_node.dbNode != ctx->slot->data.database)
704  return;
705 
706  /* output plugin doesn't look for this origin, no need to queue */
707  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
708  return;
709 
710  change = ReorderBufferGetChange(ctx->reorder);
712  change->origin_id = XLogRecGetOrigin(r);
713  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
714 
716  {
717  Size datalen;
718  Size tuplelen;
719 
720  data = XLogRecGetBlockData(r, 0, &datalen);
721 
722  tuplelen = datalen - SizeOfHeapHeader;
723 
724  change->data.tp.newtuple =
725  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
726 
727  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
728  }
729 
730  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
731  {
732  Size datalen;
733  Size tuplelen;
734 
735  /* caution, remaining data in record is not aligned */
736  data = XLogRecGetData(r) + SizeOfHeapUpdate;
737  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
738  tuplelen = datalen - SizeOfHeapHeader;
739 
740  change->data.tp.oldtuple =
741  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
742 
743  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
744  }
745 
746  change->data.tp.clear_toast_afterwards = true;
747 
748  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
749 }
struct ReorderBuffer * reorder
Definition: logical.h:45
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:954
union ReorderBufferChange::@93 data
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:115
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:477
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:77
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:219
#define XLH_UPDATE_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:80
#define SizeOfHeapUpdate
Definition: heapam_xlog.h:203
#define XLogRecGetData(decoder)
Definition: xlogreader.h:220
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:221
struct ReorderBufferChange::@93::@94 tp
XLogReaderState * record
Definition: decode.c:53
RepOriginId origin_id
Definition: reorderbuffer.h:79
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1307
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1331
#define NULL
Definition: c.h:229
size_t Size
Definition: c.h:356
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:51
ReplicationSlot * slot
Definition: logical.h:41
#define XLH_UPDATE_CONTAINS_OLD
Definition: heapam_xlog.h:85
#define SizeOfHeapHeader
Definition: heapam_xlog.h:127
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
static void DecodeXactOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 208 of file decode.c.

References DecodeAbort(), DecodeCommit(), elog, ERROR, i, xl_xact_assignment::nsubxacts, XLogRecordBuffer::origptr, ParseAbortRecord(), ParseCommitRecord(), XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAssignChild(), ReorderBufferProcessXid(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), LogicalDecodingContext::snapshot_builder, TransactionIdIsValid, 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_OPMASK, XLOG_XACT_PREPARE, XLogRecGetData, XLogRecGetInfo, XLogRecGetXid, xl_xact_assignment::xsub, and xl_xact_assignment::xtop.

Referenced by LogicalDecodingProcessRecord().

209 {
210  SnapBuild *builder = ctx->snapshot_builder;
211  ReorderBuffer *reorder = ctx->reorder;
212  XLogReaderState *r = buf->record;
214 
215  /*
216  * No point in doing anything yet, data could not be decoded anyway. It's
217  * ok not to call ReorderBufferProcessXid() in that case, except in the
218  * assignment case there'll not be any later records with the same xid;
219  * and in the assignment case we'll not decode those xacts.
220  */
222  return;
223 
224  switch (info)
225  {
226  case XLOG_XACT_COMMIT:
228  {
229  xl_xact_commit *xlrec;
230  xl_xact_parsed_commit parsed;
231  TransactionId xid;
232 
233  xlrec = (xl_xact_commit *) XLogRecGetData(r);
234  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
235 
236  if (!TransactionIdIsValid(parsed.twophase_xid))
237  xid = XLogRecGetXid(r);
238  else
239  xid = parsed.twophase_xid;
240 
241  DecodeCommit(ctx, buf, &parsed, xid);
242  break;
243  }
244  case XLOG_XACT_ABORT:
246  {
247  xl_xact_abort *xlrec;
248  xl_xact_parsed_abort parsed;
249  TransactionId xid;
250 
251  xlrec = (xl_xact_abort *) XLogRecGetData(r);
252  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
253 
254  if (!TransactionIdIsValid(parsed.twophase_xid))
255  xid = XLogRecGetXid(r);
256  else
257  xid = parsed.twophase_xid;
258 
259  DecodeAbort(ctx, buf, &parsed, xid);
260  break;
261  }
263  {
264  xl_xact_assignment *xlrec;
265  int i;
266  TransactionId *sub_xid;
267 
268  xlrec = (xl_xact_assignment *) XLogRecGetData(r);
269 
270  sub_xid = &xlrec->xsub[0];
271 
272  for (i = 0; i < xlrec->nsubxacts; i++)
273  {
274  ReorderBufferAssignChild(reorder, xlrec->xtop,
275  *(sub_xid++), buf->origptr);
276  }
277  break;
278  }
279  case XLOG_XACT_PREPARE:
280 
281  /*
282  * Currently decoding ignores PREPARE TRANSACTION and will just
283  * decode the transaction when the COMMIT PREPARED is sent or
284  * throw away the transaction's contents when a ROLLBACK PREPARED
285  * is received. In the future we could add code to expose prepared
286  * transactions in the changestream allowing for a kind of
287  * distributed 2PC.
288  */
289  ReorderBufferProcessXid(reorder, XLogRecGetXid(r), buf->origptr);
290  break;
291  default:
292  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
293  }
294 }
#define XLOG_XACT_COMMIT
Definition: xact.h:134
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid)
Definition: decode.c:536
struct ReorderBuffer * reorder
Definition: logical.h:45
#define XLOG_XACT_PREPARE
Definition: xact.h:135
uint32 TransactionId
Definition: c.h:397
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:401
TransactionId twophase_xid
Definition: xact.h:305
unsigned char uint8
Definition: c.h:266
TransactionId xtop
Definition: xact.h:183
TransactionId xsub[FLEXIBLE_ARRAY_MEMBER]
Definition: xact.h:185
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid)
Definition: decode.c:620
#define XLogRecGetData(decoder)
Definition: xlogreader.h:220
#define ERROR
Definition: elog.h:43
void ParseCommitRecord(uint8 info, xl_xact_commit *xlrec, xl_xact_parsed_commit *parsed)
Definition: xactdesc.c:35
#define XLOG_XACT_ABORT_PREPARED
Definition: xact.h:138
XLogReaderState * record
Definition: decode.c:53
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:216
void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:139
TransactionId twophase_xid
Definition: xact.h:322
struct SnapBuild * snapshot_builder
Definition: logical.h:46
#define XLOG_XACT_ABORT
Definition: xact.h:136
#define XLOG_XACT_OPMASK
Definition: xact.h:144
void ParseAbortRecord(uint8 info, xl_xact_abort *xlrec, xl_xact_parsed_abort *parsed)
Definition: xactdesc.c:122
XLogRecPtr origptr
Definition: decode.c:51
int i
#define elog
Definition: elog.h:219
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:137
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
static void DecodeXLogOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 166 of file decode.c.

References elog, ERROR, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SnapBuildSerializationPoint(), LogicalDecodingContext::snapshot_builder, 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_PARAMETER_CHANGE, XLOG_RESTORE_POINT, XLOG_SWITCH, XLogRecGetInfo, XLogRecGetXid, and XLR_INFO_MASK.

Referenced by LogicalDecodingProcessRecord().

167 {
168  SnapBuild *builder = ctx->snapshot_builder;
169  uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
170 
172  buf->origptr);
173 
174  switch (info)
175  {
176  /* this is also used in END_OF_RECOVERY checkpoints */
179  SnapBuildSerializationPoint(builder, buf->origptr);
180 
181  break;
183 
184  /*
185  * a RUNNING_XACTS record will have been logged near to this, we
186  * can restart from there.
187  */
188  break;
189  case XLOG_NOOP:
190  case XLOG_NEXTOID:
191  case XLOG_SWITCH:
192  case XLOG_BACKUP_END:
194  case XLOG_RESTORE_POINT:
195  case XLOG_FPW_CHANGE:
196  case XLOG_FPI_FOR_HINT:
197  case XLOG_FPI:
198  break;
199  default:
200  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
201  }
202 }
#define XLOG_FPW_CHANGE
Definition: pg_control.h:74
struct ReorderBuffer * reorder
Definition: logical.h:45
#define XLOG_CHECKPOINT_ONLINE
Definition: pg_control.h:67
unsigned char uint8
Definition: c.h:266
#define XLOG_RESTORE_POINT
Definition: pg_control.h:73
#define XLOG_END_OF_RECOVERY
Definition: pg_control.h:75
#define XLOG_NEXTOID
Definition: pg_control.h:69
#define ERROR
Definition: elog.h:43
#define XLOG_PARAMETER_CHANGE
Definition: pg_control.h:72
#define XLOG_CHECKPOINT_SHUTDOWN
Definition: pg_control.h:66
XLogReaderState * record
Definition: decode.c:53
#define XLOG_BACKUP_END
Definition: pg_control.h:71
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:216
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
#define XLOG_NOOP
Definition: pg_control.h:68
struct SnapBuild * snapshot_builder
Definition: logical.h:46
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
#define XLOG_FPI_FOR_HINT
Definition: pg_control.h:76
#define XLOG_SWITCH
Definition: pg_control.h:70
XLogRecPtr origptr
Definition: decode.c:51
#define XLOG_FPI
Definition: pg_control.h:77
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1464
#define elog
Definition: elog.h:219
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
static void DecodeXLogTuple ( char *  data,
Size  len,
ReorderBufferTupleBuf tup 
)
static

Definition at line 954 of file decode.c.

References Assert, header(), InvalidOid, ItemPointerSetInvalid, 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, HeapTupleData::t_tableOid, and ReorderBufferTupleBuf::tuple.

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

955 {
956  xl_heap_header xlhdr;
957  int datalen = len - SizeOfHeapHeader;
959 
960  Assert(datalen >= 0);
961 
962  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
963  header = tuple->tuple.t_data;
964 
965  /* not a disk based tuple */
966  ItemPointerSetInvalid(&tuple->tuple.t_self);
967 
968  /* we can only figure this out after reassembling the transactions */
969  tuple->tuple.t_tableOid = InvalidOid;
970 
971  /* data is not stored aligned, copy to aligned storage */
972  memcpy((char *) &xlhdr,
973  data,
974  SizeOfHeapHeader);
975 
976  memset(header, 0, SizeofHeapTupleHeader);
977 
978  memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
979  data + SizeOfHeapHeader,
980  datalen);
981 
982  header->t_infomask = xlhdr.t_infomask;
983  header->t_infomask2 = xlhdr.t_infomask2;
984  header->t_hoff = xlhdr.t_hoff;
985 }
#define SizeofHeapTupleHeader
Definition: htup_details.h:170
uint16 t_infomask2
Definition: heapam_xlog.h:122
#define InvalidOid
Definition: postgres_ext.h:36
#define Assert(condition)
Definition: c.h:675
uint16 t_infomask
Definition: heapam_xlog.h:123
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:207
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:149
#define SizeOfHeapHeader
Definition: heapam_xlog.h:127
static bool FilterByOrigin ( LogicalDecodingContext ctx,
RepOriginId  origin_id 
)
inlinestatic

Definition at line 477 of file decode.c.

References LogicalDecodingContext::callbacks, OutputPluginCallbacks::filter_by_origin_cb, filter_by_origin_cb_wrapper(), and NULL.

Referenced by DecodeCommit(), DecodeDelete(), DecodeInsert(), DecodeLogicalMsgOp(), DecodeMultiInsert(), DecodeSpecConfirm(), and DecodeUpdate().

478 {
479  if (ctx->callbacks.filter_by_origin_cb == NULL)
480  return false;
481 
482  return filter_by_origin_cb_wrapper(ctx, origin_id);
483 }
OutputPluginCallbacks callbacks
Definition: logical.h:48
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:718
#define NULL
Definition: c.h:229
LogicalDecodeFilterByOriginCB filter_by_origin_cb
void LogicalDecodingProcessRecord ( LogicalDecodingContext ctx,
XLogReaderState record 
)

Definition at line 93 of file decode.c.

References buf, DecodeHeap2Op(), DecodeHeapOp(), DecodeLogicalMsgOp(), DecodeStandbyOp(), DecodeXactOp(), DecodeXLogOp(), elog, XLogRecordBuffer::endptr, XLogReaderState::EndRecPtr, ERROR, XLogRecordBuffer::origptr, LogicalDecodingContext::reader, XLogReaderState::ReadRecPtr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), RM_NEXT_ID, XLogRecGetRmid, and XLogRecGetXid.

Referenced by DecodingContextFindStartpoint(), pg_logical_slot_get_changes_guts(), and XLogSendLogical().

94 {
96 
97  buf.origptr = ctx->reader->ReadRecPtr;
98  buf.endptr = ctx->reader->EndRecPtr;
99  buf.record = record;
100 
101  /* cast so we get a warning when new rmgrs are added */
102  switch ((RmgrIds) XLogRecGetRmid(record))
103  {
104  /*
105  * Rmgrs we care about for logical decoding. Add new rmgrs in
106  * rmgrlist.h's order.
107  */
108  case RM_XLOG_ID:
109  DecodeXLogOp(ctx, &buf);
110  break;
111 
112  case RM_XACT_ID:
113  DecodeXactOp(ctx, &buf);
114  break;
115 
116  case RM_STANDBY_ID:
117  DecodeStandbyOp(ctx, &buf);
118  break;
119 
120  case RM_HEAP2_ID:
121  DecodeHeap2Op(ctx, &buf);
122  break;
123 
124  case RM_HEAP_ID:
125  DecodeHeapOp(ctx, &buf);
126  break;
127 
128  case RM_LOGICALMSG_ID:
129  DecodeLogicalMsgOp(ctx, &buf);
130  break;
131 
132  /*
133  * Rmgrs irrelevant for logical decoding; they describe stuff not
134  * represented in logical decoding. Add new rmgrs in rmgrlist.h's
135  * order.
136  */
137  case RM_SMGR_ID:
138  case RM_CLOG_ID:
139  case RM_DBASE_ID:
140  case RM_TBLSPC_ID:
141  case RM_MULTIXACT_ID:
142  case RM_RELMAP_ID:
143  case RM_BTREE_ID:
144  case RM_HASH_ID:
145  case RM_GIN_ID:
146  case RM_GIST_ID:
147  case RM_SEQ_ID:
148  case RM_SPGIST_ID:
149  case RM_BRIN_ID:
150  case RM_COMMIT_TS_ID:
151  case RM_REPLORIGIN_ID:
152  case RM_GENERIC_ID:
153  /* just deal with xid, and done */
155  buf.origptr);
156  break;
157  case RM_NEXT_ID:
158  elog(ERROR, "unexpected RM_NEXT_ID rmgr_id: %u", (RmgrIds) XLogRecGetRmid(buf.record));
159  }
160 }
struct ReorderBuffer * reorder
Definition: logical.h:45
XLogRecPtr EndRecPtr
Definition: xlogreader.h:115
static void DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:208
static void DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:489
#define ERROR
Definition: elog.h:43
XLogRecPtr endptr
Definition: decode.c:52
static char * buf
Definition: pg_test_fsync.c:66
XLogRecPtr ReadRecPtr
Definition: xlogreader.h:114
XLogReaderState * record
Definition: decode.c:53
static void DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:166
static void DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:348
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:218
XLogRecPtr origptr
Definition: decode.c:51
XLogReaderState * reader
Definition: logical.h:44
static void DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:403
#define elog
Definition: elog.h:219
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
RmgrIds
Definition: rmgr.h:25
static void DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:300
#define XLogRecGetRmid(decoder)
Definition: xlogreader.h:217