PostgreSQL Source Code  git master
decode.c
Go to the documentation of this file.
1 /* -------------------------------------------------------------------------
2  *
3  * decode.c
4  * This module decodes WAL records read using xlogreader.h's APIs for the
5  * purpose of logical decoding by passing information to the
6  * reorderbuffer module (containing the actual changes) and to the
7  * snapbuild module to build a fitting catalog snapshot (to be able to
8  * properly decode the changes in the reorderbuffer).
9  *
10  * NOTE:
11  * This basically tries to handle all low level xlog stuff for
12  * reorderbuffer.c and snapbuild.c. There's some minor leakage where a
13  * specific record's struct is used to pass data along, but those just
14  * happen to contain the right amount of data in a convenient
15  * format. There isn't and shouldn't be much intelligence about the
16  * contents of records in here except turning them into a more usable
17  * format.
18  *
19  * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
20  * Portions Copyright (c) 1994, Regents of the University of California
21  *
22  * IDENTIFICATION
23  * src/backend/replication/logical/decode.c
24  *
25  * -------------------------------------------------------------------------
26  */
27 #include "postgres.h"
28 
29 #include "access/heapam.h"
30 #include "access/heapam_xlog.h"
31 #include "access/transam.h"
32 #include "access/xact.h"
33 #include "access/xlog_internal.h"
34 #include "access/xlogutils.h"
35 #include "access/xlogreader.h"
36 #include "access/xlogrecord.h"
37 
38 #include "catalog/pg_control.h"
39 
40 #include "replication/decode.h"
41 #include "replication/logical.h"
42 #include "replication/message.h"
44 #include "replication/origin.h"
45 #include "replication/snapbuild.h"
46 
47 #include "storage/standby.h"
48 
49 typedef struct XLogRecordBuffer
50 {
55 
56 /* RMGR Handlers */
63 
64 /* individual record(group)'s handlers */
71 
75  xl_xact_parsed_abort *parsed, TransactionId xid);
76 
77 /* common function to decode tuples */
78 static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup);
79 
80 /*
81  * Take every XLogReadRecord()ed record and perform the actions required to
82  * decode it using the output plugin already setup in the logical decoding
83  * context.
84  *
85  * NB: Note that every record's xid needs to be processed by reorderbuffer
86  * (xids contained in the content of records are not relevant for this rule).
87  * That means that for records which'd otherwise not go through the
88  * reorderbuffer ReorderBufferProcessXid() has to be called. We don't want to
89  * call ReorderBufferProcessXid for each record type by default, because
90  * e.g. empty xacts can be handled more efficiently if there's no previous
91  * state for them.
92  *
93  * We also support the ability to fast forward thru records, skipping some
94  * record types completely - see individual record types for details.
95  */
96 void
98 {
100 
101  buf.origptr = ctx->reader->ReadRecPtr;
102  buf.endptr = ctx->reader->EndRecPtr;
103  buf.record = record;
104 
105  /* cast so we get a warning when new rmgrs are added */
106  switch ((RmgrIds) XLogRecGetRmid(record))
107  {
108  /*
109  * Rmgrs we care about for logical decoding. Add new rmgrs in
110  * rmgrlist.h's order.
111  */
112  case RM_XLOG_ID:
113  DecodeXLogOp(ctx, &buf);
114  break;
115 
116  case RM_XACT_ID:
117  DecodeXactOp(ctx, &buf);
118  break;
119 
120  case RM_STANDBY_ID:
121  DecodeStandbyOp(ctx, &buf);
122  break;
123 
124  case RM_HEAP2_ID:
125  DecodeHeap2Op(ctx, &buf);
126  break;
127 
128  case RM_HEAP_ID:
129  DecodeHeapOp(ctx, &buf);
130  break;
131 
132  case RM_LOGICALMSG_ID:
133  DecodeLogicalMsgOp(ctx, &buf);
134  break;
135 
136  /*
137  * Rmgrs irrelevant for logical decoding; they describe stuff not
138  * represented in logical decoding. Add new rmgrs in rmgrlist.h's
139  * order.
140  */
141  case RM_SMGR_ID:
142  case RM_CLOG_ID:
143  case RM_DBASE_ID:
144  case RM_TBLSPC_ID:
145  case RM_MULTIXACT_ID:
146  case RM_RELMAP_ID:
147  case RM_BTREE_ID:
148  case RM_HASH_ID:
149  case RM_GIN_ID:
150  case RM_GIST_ID:
151  case RM_SEQ_ID:
152  case RM_SPGIST_ID:
153  case RM_BRIN_ID:
154  case RM_COMMIT_TS_ID:
155  case RM_REPLORIGIN_ID:
156  case RM_GENERIC_ID:
157  /* just deal with xid, and done */
159  buf.origptr);
160  break;
161  case RM_NEXT_ID:
162  elog(ERROR, "unexpected RM_NEXT_ID rmgr_id: %u", (RmgrIds) XLogRecGetRmid(buf.record));
163  }
164 }
165 
166 /*
167  * Handle rmgr XLOG_ID records for DecodeRecordIntoReorderBuffer().
168  */
169 static void
171 {
172  SnapBuild *builder = ctx->snapshot_builder;
173  uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
174 
176  buf->origptr);
177 
178  switch (info)
179  {
180  /* this is also used in END_OF_RECOVERY checkpoints */
183  SnapBuildSerializationPoint(builder, buf->origptr);
184 
185  break;
187 
188  /*
189  * a RUNNING_XACTS record will have been logged near to this, we
190  * can restart from there.
191  */
192  break;
193  case XLOG_NOOP:
194  case XLOG_NEXTOID:
195  case XLOG_SWITCH:
196  case XLOG_BACKUP_END:
198  case XLOG_RESTORE_POINT:
199  case XLOG_FPW_CHANGE:
200  case XLOG_FPI_FOR_HINT:
201  case XLOG_FPI:
202  break;
203  default:
204  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
205  }
206 }
207 
208 /*
209  * Handle rmgr XACT_ID records for DecodeRecordIntoReorderBuffer().
210  */
211 static void
213 {
214  SnapBuild *builder = ctx->snapshot_builder;
215  ReorderBuffer *reorder = ctx->reorder;
216  XLogReaderState *r = buf->record;
218 
219  /*
220  * If the snapshot isn't yet fully built, we cannot decode anything, so
221  * bail out.
222  *
223  * However, it's critical to process XLOG_XACT_ASSIGNMENT records even
224  * when the snapshot is being built: it is possible to get later records
225  * that require subxids to be properly assigned.
226  */
228  info != XLOG_XACT_ASSIGNMENT)
229  return;
230 
231  switch (info)
232  {
233  case XLOG_XACT_COMMIT:
235  {
236  xl_xact_commit *xlrec;
237  xl_xact_parsed_commit parsed;
238  TransactionId xid;
239 
240  xlrec = (xl_xact_commit *) XLogRecGetData(r);
241  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
242 
243  if (!TransactionIdIsValid(parsed.twophase_xid))
244  xid = XLogRecGetXid(r);
245  else
246  xid = parsed.twophase_xid;
247 
248  DecodeCommit(ctx, buf, &parsed, xid);
249  break;
250  }
251  case XLOG_XACT_ABORT:
253  {
254  xl_xact_abort *xlrec;
255  xl_xact_parsed_abort parsed;
256  TransactionId xid;
257 
258  xlrec = (xl_xact_abort *) XLogRecGetData(r);
259  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
260 
261  if (!TransactionIdIsValid(parsed.twophase_xid))
262  xid = XLogRecGetXid(r);
263  else
264  xid = parsed.twophase_xid;
265 
266  DecodeAbort(ctx, buf, &parsed, xid);
267  break;
268  }
270  {
271  xl_xact_assignment *xlrec;
272  int i;
273  TransactionId *sub_xid;
274 
275  xlrec = (xl_xact_assignment *) XLogRecGetData(r);
276 
277  sub_xid = &xlrec->xsub[0];
278 
279  for (i = 0; i < xlrec->nsubxacts; i++)
280  {
281  ReorderBufferAssignChild(reorder, xlrec->xtop,
282  *(sub_xid++), buf->origptr);
283  }
284  break;
285  }
286  case XLOG_XACT_PREPARE:
287 
288  /*
289  * Currently decoding ignores PREPARE TRANSACTION and will just
290  * decode the transaction when the COMMIT PREPARED is sent or
291  * throw away the transaction's contents when a ROLLBACK PREPARED
292  * is received. In the future we could add code to expose prepared
293  * transactions in the changestream allowing for a kind of
294  * distributed 2PC.
295  */
296  ReorderBufferProcessXid(reorder, XLogRecGetXid(r), buf->origptr);
297  break;
298  default:
299  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
300  }
301 }
302 
303 /*
304  * Handle rmgr STANDBY_ID records for DecodeRecordIntoReorderBuffer().
305  */
306 static void
308 {
309  SnapBuild *builder = ctx->snapshot_builder;
310  XLogReaderState *r = buf->record;
311  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
312 
314 
315  switch (info)
316  {
317  case XLOG_RUNNING_XACTS:
318  {
320 
321  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
322 
323  /*
324  * Abort all transactions that we keep track of, that are
325  * older than the record's oldestRunningXid. This is the most
326  * convenient spot for doing so since, in contrast to shutdown
327  * or end-of-recovery checkpoints, we have information about
328  * all running transactions which includes prepared ones,
329  * while shutdown checkpoints just know that no non-prepared
330  * transactions are in progress.
331  */
333  }
334  break;
335  case XLOG_STANDBY_LOCK:
336  break;
337  case XLOG_INVALIDATIONS:
338  {
339  xl_invalidations *invalidations =
341 
342  if (!ctx->fast_forward)
344  invalidations->nmsgs,
345  invalidations->msgs);
346  }
347  break;
348  default:
349  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
350  }
351 }
352 
353 /*
354  * Handle rmgr HEAP2_ID records for DecodeRecordIntoReorderBuffer().
355  */
356 static void
358 {
360  TransactionId xid = XLogRecGetXid(buf->record);
361  SnapBuild *builder = ctx->snapshot_builder;
362 
363  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
364 
365  /*
366  * If we don't have snapshot or we are just fast-forwarding, there is no
367  * point in decoding changes.
368  */
370  ctx->fast_forward)
371  return;
372 
373  switch (info)
374  {
376  if (!ctx->fast_forward &&
377  SnapBuildProcessChange(builder, xid, buf->origptr))
378  DecodeMultiInsert(ctx, buf);
379  break;
380  case XLOG_HEAP2_NEW_CID:
381  {
382  xl_heap_new_cid *xlrec;
383 
384  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
385  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
386 
387  break;
388  }
389  case XLOG_HEAP2_REWRITE:
390 
391  /*
392  * Although these records only exist to serve the needs of logical
393  * decoding, all the work happens as part of crash or archive
394  * recovery, so we don't need to do anything here.
395  */
396  break;
397 
398  /*
399  * Everything else here is just low level physical stuff we're not
400  * interested in.
401  */
403  case XLOG_HEAP2_CLEAN:
405  case XLOG_HEAP2_VISIBLE:
407  break;
408  default:
409  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
410  }
411 }
412 
413 /*
414  * Handle rmgr HEAP_ID records for DecodeRecordIntoReorderBuffer().
415  */
416 static void
418 {
420  TransactionId xid = XLogRecGetXid(buf->record);
421  SnapBuild *builder = ctx->snapshot_builder;
422 
423  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
424 
425  /*
426  * If we don't have snapshot or we are just fast-forwarding, there is no
427  * point in decoding data changes.
428  */
430  ctx->fast_forward)
431  return;
432 
433  switch (info)
434  {
435  case XLOG_HEAP_INSERT:
436  if (SnapBuildProcessChange(builder, xid, buf->origptr))
437  DecodeInsert(ctx, buf);
438  break;
439 
440  /*
441  * Treat HOT update as normal updates. There is no useful
442  * information in the fact that we could make it a HOT update
443  * locally and the WAL layout is compatible.
444  */
446  case XLOG_HEAP_UPDATE:
447  if (SnapBuildProcessChange(builder, xid, buf->origptr))
448  DecodeUpdate(ctx, buf);
449  break;
450 
451  case XLOG_HEAP_DELETE:
452  if (SnapBuildProcessChange(builder, xid, buf->origptr))
453  DecodeDelete(ctx, buf);
454  break;
455 
456  case XLOG_HEAP_TRUNCATE:
457  if (SnapBuildProcessChange(builder, xid, buf->origptr))
458  DecodeTruncate(ctx, buf);
459  break;
460 
461  case XLOG_HEAP_INPLACE:
462 
463  /*
464  * Inplace updates are only ever performed on catalog tuples and
465  * can, per definition, not change tuple visibility. Since we
466  * don't decode catalog tuples, we're not interested in the
467  * record's contents.
468  *
469  * In-place updates can be used either by XID-bearing transactions
470  * (e.g. in CREATE INDEX CONCURRENTLY) or by XID-less
471  * transactions (e.g. VACUUM). In the former case, the commit
472  * record will include cache invalidations, so we mark the
473  * transaction as catalog modifying here. Currently that's
474  * redundant because the commit will do that as well, but once we
475  * support decoding in-progress relations, this will be important.
476  */
477  if (!TransactionIdIsValid(xid))
478  break;
479 
480  SnapBuildProcessChange(builder, xid, buf->origptr);
482  break;
483 
484  case XLOG_HEAP_CONFIRM:
485  if (SnapBuildProcessChange(builder, xid, buf->origptr))
486  DecodeSpecConfirm(ctx, buf);
487  break;
488 
489  case XLOG_HEAP_LOCK:
490  /* we don't care about row level locks for now */
491  break;
492 
493  default:
494  elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
495  break;
496  }
497 }
498 
499 static inline bool
501 {
502  if (ctx->callbacks.filter_by_origin_cb == NULL)
503  return false;
504 
505  return filter_by_origin_cb_wrapper(ctx, origin_id);
506 }
507 
508 /*
509  * Handle rmgr LOGICALMSG_ID records for DecodeRecordIntoReorderBuffer().
510  */
511 static void
513 {
514  SnapBuild *builder = ctx->snapshot_builder;
515  XLogReaderState *r = buf->record;
516  TransactionId xid = XLogRecGetXid(r);
517  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
518  RepOriginId origin_id = XLogRecGetOrigin(r);
519  Snapshot snapshot;
520  xl_logical_message *message;
521 
522  if (info != XLOG_LOGICAL_MESSAGE)
523  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
524 
526 
527  /*
528  * If we don't have snapshot or we are just fast-forwarding, there is no
529  * point in decoding messages.
530  */
532  ctx->fast_forward)
533  return;
534 
535  message = (xl_logical_message *) XLogRecGetData(r);
536 
537  if (message->dbId != ctx->slot->data.database ||
538  FilterByOrigin(ctx, origin_id))
539  return;
540 
541  if (message->transactional &&
542  !SnapBuildProcessChange(builder, xid, buf->origptr))
543  return;
544  else if (!message->transactional &&
546  SnapBuildXactNeedsSkip(builder, buf->origptr)))
547  return;
548 
549  snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
550  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
551  message->transactional,
552  message->message, /* first part of message is
553  * prefix */
554  message->message_size,
555  message->message + message->prefix_size);
556 }
557 
558 /*
559  * Consolidated commit record handling between the different form of commit
560  * records.
561  */
562 static void
565 {
566  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
567  TimestampTz commit_time = parsed->xact_time;
568  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
569  int i;
570 
571  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
572  {
573  origin_lsn = parsed->origin_lsn;
574  commit_time = parsed->origin_timestamp;
575  }
576 
577  /*
578  * Process invalidation messages, even if we're not interested in the
579  * transaction's contents, since the various caches need to always be
580  * consistent.
581  */
582  if (parsed->nmsgs > 0)
583  {
584  if (!ctx->fast_forward)
586  parsed->nmsgs, parsed->msgs);
588  }
589 
591  parsed->nsubxacts, parsed->subxacts);
592 
593  /* ----
594  * Check whether we are interested in this specific transaction, and tell
595  * the reorderbuffer to forget the content of the (sub-)transactions
596  * if not.
597  *
598  * There can be several reasons we might not be interested in this
599  * transaction:
600  * 1) We might not be interested in decoding transactions up to this
601  * LSN. This can happen because we previously decoded it and now just
602  * are restarting or if we haven't assembled a consistent snapshot yet.
603  * 2) The transaction happened in another database.
604  * 3) The output plugin is not interested in the origin.
605  * 4) We are doing fast-forwarding
606  *
607  * We can't just use ReorderBufferAbort() here, because we need to execute
608  * the transaction's invalidations. This currently won't be needed if
609  * we're just skipping over the transaction because currently we only do
610  * so during startup, to get to the first transaction the client needs. As
611  * we have reset the catalog caches before starting to read WAL, and we
612  * haven't yet touched any catalogs, there can't be anything to invalidate.
613  * But if we're "forgetting" this commit because it's it happened in
614  * another database, the invalidations might be important, because they
615  * could be for shared catalogs and we might have loaded data into the
616  * relevant syscaches.
617  * ---
618  */
620  (parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
621  ctx->fast_forward || FilterByOrigin(ctx, origin_id))
622  {
623  for (i = 0; i < parsed->nsubxacts; i++)
624  {
625  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
626  }
627  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
628 
629  return;
630  }
631 
632  /* tell the reorderbuffer about the surviving subtransactions */
633  for (i = 0; i < parsed->nsubxacts; i++)
634  {
635  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
636  buf->origptr, buf->endptr);
637  }
638 
639  /* replay actions of all transaction + subtransactions in order */
640  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
641  commit_time, origin_id, origin_lsn);
642 }
643 
644 /*
645  * Get the data from the various forms of abort records and pass it on to
646  * snapbuild.c and reorderbuffer.c
647  */
648 static void
650  xl_xact_parsed_abort *parsed, TransactionId xid)
651 {
652  int i;
653 
654  for (i = 0; i < parsed->nsubxacts; i++)
655  {
656  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
657  buf->record->EndRecPtr);
658  }
659 
660  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
661 }
662 
663 /*
664  * Parse XLOG_HEAP_INSERT (not MULTI_INSERT!) records into tuplebufs.
665  *
666  * Deletes can contain the new tuple.
667  */
668 static void
670 {
671  Size datalen;
672  char *tupledata;
673  Size tuplelen;
674  XLogReaderState *r = buf->record;
675  xl_heap_insert *xlrec;
676  ReorderBufferChange *change;
677  RelFileNode target_node;
678 
679  xlrec = (xl_heap_insert *) XLogRecGetData(r);
680 
681  /*
682  * Ignore insert records without new tuples (this does happen when
683  * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
684  */
685  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
686  return;
687 
688  /* only interested in our database */
689  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
690  if (target_node.dbNode != ctx->slot->data.database)
691  return;
692 
693  /* output plugin doesn't look for this origin, no need to queue */
694  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
695  return;
696 
697  change = ReorderBufferGetChange(ctx->reorder);
698  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
700  else
702  change->origin_id = XLogRecGetOrigin(r);
703 
704  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
705 
706  tupledata = XLogRecGetBlockData(r, 0, &datalen);
707  tuplelen = datalen - SizeOfHeapHeader;
708 
709  change->data.tp.newtuple =
710  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
711 
712  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
713 
714  change->data.tp.clear_toast_afterwards = true;
715 
716  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
717 }
718 
719 /*
720  * Parse XLOG_HEAP_UPDATE and XLOG_HEAP_HOT_UPDATE, which have the same layout
721  * in the record, from wal into proper tuplebufs.
722  *
723  * Updates can possibly contain a new tuple and the old primary key.
724  */
725 static void
727 {
728  XLogReaderState *r = buf->record;
729  xl_heap_update *xlrec;
730  ReorderBufferChange *change;
731  char *data;
732  RelFileNode target_node;
733 
734  xlrec = (xl_heap_update *) XLogRecGetData(r);
735 
736  /* only interested in our database */
737  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
738  if (target_node.dbNode != ctx->slot->data.database)
739  return;
740 
741  /* output plugin doesn't look for this origin, no need to queue */
742  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
743  return;
744 
745  change = ReorderBufferGetChange(ctx->reorder);
747  change->origin_id = XLogRecGetOrigin(r);
748  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
749 
751  {
752  Size datalen;
753  Size tuplelen;
754 
755  data = XLogRecGetBlockData(r, 0, &datalen);
756 
757  tuplelen = datalen - SizeOfHeapHeader;
758 
759  change->data.tp.newtuple =
760  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
761 
762  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
763  }
764 
765  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
766  {
767  Size datalen;
768  Size tuplelen;
769 
770  /* caution, remaining data in record is not aligned */
771  data = XLogRecGetData(r) + SizeOfHeapUpdate;
772  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
773  tuplelen = datalen - SizeOfHeapHeader;
774 
775  change->data.tp.oldtuple =
776  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
777 
778  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
779  }
780 
781  change->data.tp.clear_toast_afterwards = true;
782 
783  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
784 }
785 
786 /*
787  * Parse XLOG_HEAP_DELETE from wal into proper tuplebufs.
788  *
789  * Deletes can possibly contain the old primary key.
790  */
791 static void
793 {
794  XLogReaderState *r = buf->record;
795  xl_heap_delete *xlrec;
796  ReorderBufferChange *change;
797  RelFileNode target_node;
798 
799  xlrec = (xl_heap_delete *) XLogRecGetData(r);
800 
801  /* only interested in our database */
802  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
803  if (target_node.dbNode != ctx->slot->data.database)
804  return;
805 
806  /*
807  * Super deletions are irrelevant for logical decoding, it's driven by the
808  * confirmation records.
809  */
810  if (xlrec->flags & XLH_DELETE_IS_SUPER)
811  return;
812 
813  /* output plugin doesn't look for this origin, no need to queue */
814  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
815  return;
816 
817  change = ReorderBufferGetChange(ctx->reorder);
819  change->origin_id = XLogRecGetOrigin(r);
820 
821  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
822 
823  /* old primary key stored */
824  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
825  {
826  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
827  Size tuplelen = datalen - SizeOfHeapHeader;
828 
829  Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
830 
831  change->data.tp.oldtuple =
832  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
833 
834  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
835  datalen, change->data.tp.oldtuple);
836  }
837 
838  change->data.tp.clear_toast_afterwards = true;
839 
840  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
841 }
842 
843 /*
844  * Parse XLOG_HEAP_TRUNCATE from wal
845  */
846 static void
848 {
849  XLogReaderState *r = buf->record;
850  xl_heap_truncate *xlrec;
851  ReorderBufferChange *change;
852 
853  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
854 
855  /* only interested in our database */
856  if (xlrec->dbId != ctx->slot->data.database)
857  return;
858 
859  /* output plugin doesn't look for this origin, no need to queue */
860  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
861  return;
862 
863  change = ReorderBufferGetChange(ctx->reorder);
865  change->origin_id = XLogRecGetOrigin(r);
866  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
867  change->data.truncate.cascade = true;
868  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
869  change->data.truncate.restart_seqs = true;
870  change->data.truncate.nrelids = xlrec->nrelids;
871  change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
872  xlrec->nrelids);
873  memcpy(change->data.truncate.relids, xlrec->relids,
874  xlrec->nrelids * sizeof(Oid));
876  buf->origptr, change);
877 }
878 
879 /*
880  * Decode XLOG_HEAP2_MULTI_INSERT_insert record into multiple tuplebufs.
881  *
882  * Currently MULTI_INSERT will always contain the full tuples.
883  */
884 static void
886 {
887  XLogReaderState *r = buf->record;
888  xl_heap_multi_insert *xlrec;
889  int i;
890  char *data;
891  char *tupledata;
892  Size tuplelen;
893  RelFileNode rnode;
894 
895  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
896 
897  /* only interested in our database */
898  XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
899  if (rnode.dbNode != ctx->slot->data.database)
900  return;
901 
902  /* output plugin doesn't look for this origin, no need to queue */
903  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
904  return;
905 
906  /*
907  * As multi_insert is not used for catalogs yet, the block should always
908  * have data even if a full-page write of it is taken.
909  */
910  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
911  Assert(tupledata != NULL);
912 
913  data = tupledata;
914  for (i = 0; i < xlrec->ntuples; i++)
915  {
916  ReorderBufferChange *change;
917  xl_multi_insert_tuple *xlhdr;
918  int datalen;
919  ReorderBufferTupleBuf *tuple;
920 
921  change = ReorderBufferGetChange(ctx->reorder);
923  change->origin_id = XLogRecGetOrigin(r);
924 
925  memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
926 
927  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
928  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
929  datalen = xlhdr->datalen;
930 
931  /*
932  * CONTAINS_NEW_TUPLE will always be set currently as multi_insert
933  * isn't used for catalogs, but better be future proof.
934  *
935  * We decode the tuple in pretty much the same way as DecodeXLogTuple,
936  * but since the layout is slightly different, we can't use it here.
937  */
939  {
941 
942  change->data.tp.newtuple =
943  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
944 
945  tuple = change->data.tp.newtuple;
946  header = tuple->tuple.t_data;
947 
948  /* not a disk based tuple */
950 
951  /*
952  * We can only figure this out after reassembling the
953  * transactions.
954  */
955  tuple->tuple.t_tableOid = InvalidOid;
956 
957  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
958 
959  memset(header, 0, SizeofHeapTupleHeader);
960 
961  memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
962  (char *) data,
963  datalen);
964  header->t_infomask = xlhdr->t_infomask;
965  header->t_infomask2 = xlhdr->t_infomask2;
966  header->t_hoff = xlhdr->t_hoff;
967  }
968 
969  /*
970  * Reset toast reassembly state only after the last row in the last
971  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
972  * call.
973  */
974  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
975  (i + 1) == xlrec->ntuples)
976  change->data.tp.clear_toast_afterwards = true;
977  else
978  change->data.tp.clear_toast_afterwards = false;
979 
981  buf->origptr, change);
982 
983  /* move to the next xl_multi_insert_tuple entry */
984  data += datalen;
985  }
986  Assert(data == tupledata + tuplelen);
987 }
988 
989 /*
990  * Parse XLOG_HEAP_CONFIRM from wal into a confirmation change.
991  *
992  * This is pretty trivial, all the state essentially already setup by the
993  * speculative insertion.
994  */
995 static void
997 {
998  XLogReaderState *r = buf->record;
999  ReorderBufferChange *change;
1000  RelFileNode target_node;
1001 
1002  /* only interested in our database */
1003  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
1004  if (target_node.dbNode != ctx->slot->data.database)
1005  return;
1006 
1007  /* output plugin doesn't look for this origin, no need to queue */
1008  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1009  return;
1010 
1011  change = ReorderBufferGetChange(ctx->reorder);
1013  change->origin_id = XLogRecGetOrigin(r);
1014 
1015  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1016 
1017  change->data.tp.clear_toast_afterwards = true;
1018 
1019  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
1020 }
1021 
1022 
1023 /*
1024  * Read a HeapTuple as WAL logged by heap_insert, heap_update and heap_delete
1025  * (but not by heap_multi_insert) into a tuplebuf.
1026  *
1027  * The size 'len' and the pointer 'data' in the record need to be
1028  * computed outside as they are record specific.
1029  */
1030 static void
1032 {
1033  xl_heap_header xlhdr;
1034  int datalen = len - SizeOfHeapHeader;
1036 
1037  Assert(datalen >= 0);
1038 
1039  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1040  header = tuple->tuple.t_data;
1041 
1042  /* not a disk based tuple */
1044 
1045  /* we can only figure this out after reassembling the transactions */
1046  tuple->tuple.t_tableOid = InvalidOid;
1047 
1048  /* data is not stored aligned, copy to aligned storage */
1049  memcpy((char *) &xlhdr,
1050  data,
1051  SizeOfHeapHeader);
1052 
1053  memset(header, 0, SizeofHeapTupleHeader);
1054 
1055  memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
1056  data + SizeOfHeapHeader,
1057  datalen);
1058 
1059  header->t_infomask = xlhdr.t_infomask;
1060  header->t_infomask2 = xlhdr.t_infomask2;
1061  header->t_hoff = xlhdr.t_hoff;
1062 }
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:885
#define XLH_TRUNCATE_CASCADE
Definition: heapam_xlog.h:116
#define SizeofHeapTupleHeader
Definition: htup_details.h:184
#define XLOG_HEAP_INSERT
Definition: heapam_xlog.h:32
void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snapshot, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define XLH_DELETE_IS_SUPER
Definition: heapam_xlog.h:95
#define XLOG_XACT_COMMIT
Definition: xact.h:143
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid)
Definition: decode.c:563
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:167
#define XLOG_FPW_CHANGE
Definition: pg_control.h:75
struct ReorderBuffer * reorder
Definition: logical.h:43
#define XLOG_XACT_PREPARE
Definition: xact.h:144
static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:669
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:59
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
uint32 TransactionId
Definition: c.h:507
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:409
#define XLOG_CHECKPOINT_ONLINE
Definition: pg_control.h:68
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:323
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:928
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:400
struct ReorderBufferChange::@101::@102 tp
int64 TimestampTz
Definition: timestamp.h:39
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:53
TransactionId twophase_xid
Definition: xact.h:317
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1031
#define XLOG_INVALIDATIONS
Definition: standbydefs.h:36
OutputPluginCallbacks callbacks
Definition: logical.h:53
unsigned char uint8
Definition: c.h:356
#define XLOG_STANDBY_LOCK
Definition: standbydefs.h:34
TransactionId * subxacts
Definition: xact.h:309
#define XLH_INSERT_IS_SPECULATIVE
Definition: heapam_xlog.h:68
uint16 RepOriginId
Definition: xlogdefs.h:58
#define XLOG_RESTORE_POINT
Definition: pg_control.h:74
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
SharedInvalidationMessage * msgs
Definition: xact.h:315
TransactionId xtop
Definition: xact.h:193
XLogRecPtr origin_lsn
Definition: xact.h:322
#define XLH_DELETE_CONTAINS_OLD
Definition: heapam_xlog.h:99
#define XLOG_HEAP_HOT_UPDATE
Definition: heapam_xlog.h:36
TransactionId xsub[FLEXIBLE_ARRAY_MEMBER]
Definition: xact.h:195
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:132
unsigned int Oid
Definition: postgres_ext.h:31
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:500
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:78
#define XLOG_END_OF_RECOVERY
Definition: pg_control.h:76
static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:726
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid)
Definition: decode.c:649
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:245
#define XLOG_HEAP2_MULTI_INSERT
Definition: heapam_xlog.h:58
#define XLH_UPDATE_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:80
#define SizeOfHeapUpdate
Definition: heapam_xlog.h:225
char message[FLEXIBLE_ARRAY_MEMBER]
Definition: message.h:26
XLogRecPtr EndRecPtr
Definition: xlogreader.h:133
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:720
HeapTupleHeader t_data
Definition: htup.h:68
#define XLOG_HEAP2_VISIBLE
Definition: heapam_xlog.h:57
static void DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:212
Oid relids[FLEXIBLE_ARRAY_MEMBER]
Definition: heapam_xlog.h:129
#define XLOG_NEXTOID
Definition: pg_control.h:70
static void DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:512
#define XLogRecGetData(decoder)
Definition: xlogreader.h:246
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:996
#define ERROR
Definition: elog.h:43
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:97
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:247
#define XLH_INSERT_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:69
#define XLOG_PARAMETER_CHANGE
Definition: pg_control.h:73
ItemPointerData t_self
Definition: htup.h:65
void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
uint32 t_len
Definition: htup.h:64
#define XLOG_HEAP2_CLEAN
Definition: heapam_xlog.h:54
#define SizeOfHeapDelete
Definition: heapam_xlog.h:111
XLogRecPtr endptr
Definition: decode.c:52
static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:792
void ParseCommitRecord(uint8 info, xl_xact_commit *xlrec, xl_xact_parsed_commit *parsed)
Definition: xactdesc.c:34
static char * buf
Definition: pg_test_fsync.c:68
#define XLOG_HEAP2_CLEANUP_INFO
Definition: heapam_xlog.h:56
#define XLOG_XACT_ABORT_PREPARED
Definition: xact.h:147
#define XLOG_CHECKPOINT_SHUTDOWN
Definition: pg_control.h:67
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:770
XLogRecPtr ReadRecPtr
Definition: xlogreader.h:132
HeapTupleData tuple
Definition: reorderbuffer.h:27
Oid t_tableOid
Definition: htup.h:66
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1095
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:60
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:805
XLogReaderState * record
Definition: decode.c:53
static void DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:170
#define XLOG_BACKUP_END
Definition: pg_control.h:72
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:242
RepOriginId origin_id
Definition: reorderbuffer.h:80
bool transactional
Definition: message.h:23
TransactionId * subxacts
Definition: xact.h:337
void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
#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:1367
static void DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:357
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:244
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:148
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1391
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
#define XLOG_NOOP
Definition: pg_control.h:69
#define XLH_TRUNCATE_RESTART_SEQS
Definition: heapam_xlog.h:117
#define XLOG_HEAP_TRUNCATE
Definition: heapam_xlog.h:35
#define InvalidOid
Definition: postgres_ext.h:36
TransactionId twophase_xid
Definition: xact.h:342
struct SnapBuild * snapshot_builder
Definition: logical.h:44
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:732
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
union ReorderBufferChange::@101 data
SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER]
Definition: standbydefs.h:69
#define XLOG_HEAP_UPDATE
Definition: heapam_xlog.h:34
#define XLOG_XACT_ABORT
Definition: xact.h:145
#define XLOG_FPI_FOR_HINT
Definition: pg_control.h:77
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
size_t Size
Definition: c.h:466
#define XLOG_XACT_OPMASK
Definition: xact.h:153
void ParseAbortRecord(uint8 info, xl_xact_abort *xlrec, xl_xact_parsed_abort *parsed)
Definition: xactdesc.c:129
#define XLOG_SWITCH
Definition: pg_control.h:71
static void DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:847
#define XLOG_HEAP_INPLACE
Definition: heapam_xlog.h:39
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:210
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
#define XLOG_HEAP2_FREEZE_PAGE
Definition: heapam_xlog.h:55
XLogRecPtr origptr
Definition: decode.c:51
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:172
#define XLOG_FPI
Definition: pg_control.h:78
XLogReaderState * reader
Definition: logical.h:42
ReplicationSlot * slot
Definition: logical.h:39
#define XLOG_HEAP_LOCK
Definition: heapam_xlog.h:38
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define elog(elevel,...)
Definition: elog.h:226
Oid * ReorderBufferGetRelids(ReorderBuffer *rb, int nrelids)
int i
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1472
#define XLOG_LOGICAL_MESSAGE
Definition: message.h:37
TransactionId oldestRunningXid
Definition: standbydefs.h:53
#define XLOG_RUNNING_XACTS
Definition: standbydefs.h:35
static void DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:417
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define XLH_UPDATE_CONTAINS_OLD
Definition: heapam_xlog.h:85
#define SHORTALIGN(LEN)
Definition: c.h:681
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:146
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
LogicalDecodeFilterByOriginCB filter_by_origin_cb
#define XLOG_HEAP_DELETE
Definition: heapam_xlog.h:33
struct XLogRecordBuffer XLogRecordBuffer
RmgrIds
Definition: rmgr.h:25
static void DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:307
#define SizeOfHeapHeader
Definition: heapam_xlog.h:149
#define XLogRecGetRmid(decoder)
Definition: xlogreader.h:243
struct ReorderBufferChange::@101::@103 truncate
#define XLOG_HEAP_CONFIRM
Definition: heapam_xlog.h:37
TimestampTz xact_time
Definition: xact.h:302
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:191
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:676
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)