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-2018, 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  * No point in doing anything yet, data could not be decoded anyway. It's
221  * ok not to call ReorderBufferProcessXid() in that case, except in the
222  * assignment case there'll not be any later records with the same xid;
223  * and in the assignment case we'll not decode those xacts.
224  */
226  return;
227 
228  switch (info)
229  {
230  case XLOG_XACT_COMMIT:
232  {
233  xl_xact_commit *xlrec;
234  xl_xact_parsed_commit parsed;
235  TransactionId xid;
236 
237  xlrec = (xl_xact_commit *) XLogRecGetData(r);
238  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
239 
240  if (!TransactionIdIsValid(parsed.twophase_xid))
241  xid = XLogRecGetXid(r);
242  else
243  xid = parsed.twophase_xid;
244 
245  DecodeCommit(ctx, buf, &parsed, xid);
246  break;
247  }
248  case XLOG_XACT_ABORT:
250  {
251  xl_xact_abort *xlrec;
252  xl_xact_parsed_abort parsed;
253  TransactionId xid;
254 
255  xlrec = (xl_xact_abort *) XLogRecGetData(r);
256  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
257 
258  if (!TransactionIdIsValid(parsed.twophase_xid))
259  xid = XLogRecGetXid(r);
260  else
261  xid = parsed.twophase_xid;
262 
263  DecodeAbort(ctx, buf, &parsed, xid);
264  break;
265  }
267  {
268  xl_xact_assignment *xlrec;
269  int i;
270  TransactionId *sub_xid;
271 
272  xlrec = (xl_xact_assignment *) XLogRecGetData(r);
273 
274  sub_xid = &xlrec->xsub[0];
275 
276  for (i = 0; i < xlrec->nsubxacts; i++)
277  {
278  ReorderBufferAssignChild(reorder, xlrec->xtop,
279  *(sub_xid++), buf->origptr);
280  }
281  break;
282  }
283  case XLOG_XACT_PREPARE:
284 
285  /*
286  * Currently decoding ignores PREPARE TRANSACTION and will just
287  * decode the transaction when the COMMIT PREPARED is sent or
288  * throw away the transaction's contents when a ROLLBACK PREPARED
289  * is received. In the future we could add code to expose prepared
290  * transactions in the changestream allowing for a kind of
291  * distributed 2PC.
292  */
293  ReorderBufferProcessXid(reorder, XLogRecGetXid(r), buf->origptr);
294  break;
295  default:
296  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
297  }
298 }
299 
300 /*
301  * Handle rmgr STANDBY_ID records for DecodeRecordIntoReorderBuffer().
302  */
303 static void
305 {
306  SnapBuild *builder = ctx->snapshot_builder;
307  XLogReaderState *r = buf->record;
308  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
309 
311 
312  switch (info)
313  {
314  case XLOG_RUNNING_XACTS:
315  {
317 
318  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
319 
320  /*
321  * Abort all transactions that we keep track of, that are
322  * older than the record's oldestRunningXid. This is the most
323  * convenient spot for doing so since, in contrast to shutdown
324  * or end-of-recovery checkpoints, we have information about
325  * all running transactions which includes prepared ones,
326  * while shutdown checkpoints just know that no non-prepared
327  * transactions are in progress.
328  */
330  }
331  break;
332  case XLOG_STANDBY_LOCK:
333  break;
334  case XLOG_INVALIDATIONS:
335  {
336  xl_invalidations *invalidations =
338 
339  if (!ctx->fast_forward)
341  invalidations->nmsgs,
342  invalidations->msgs);
343  }
344  break;
345  default:
346  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
347  }
348 }
349 
350 /*
351  * Handle rmgr HEAP2_ID records for DecodeRecordIntoReorderBuffer().
352  */
353 static void
355 {
357  TransactionId xid = XLogRecGetXid(buf->record);
358  SnapBuild *builder = ctx->snapshot_builder;
359 
360  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
361 
362  /*
363  * If we don't have snapshot or we are just fast-forwarding, there is no
364  * point in decoding changes.
365  */
367  ctx->fast_forward)
368  return;
369 
370  switch (info)
371  {
373  if (!ctx->fast_forward &&
374  SnapBuildProcessChange(builder, xid, buf->origptr))
375  DecodeMultiInsert(ctx, buf);
376  break;
377  case XLOG_HEAP2_NEW_CID:
378  {
379  xl_heap_new_cid *xlrec;
380 
381  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
382  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
383 
384  break;
385  }
386  case XLOG_HEAP2_REWRITE:
387 
388  /*
389  * Although these records only exist to serve the needs of logical
390  * decoding, all the work happens as part of crash or archive
391  * recovery, so we don't need to do anything here.
392  */
393  break;
394 
395  /*
396  * Everything else here is just low level physical stuff we're not
397  * interested in.
398  */
400  case XLOG_HEAP2_CLEAN:
402  case XLOG_HEAP2_VISIBLE:
404  break;
405  default:
406  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
407  }
408 }
409 
410 /*
411  * Handle rmgr HEAP_ID records for DecodeRecordIntoReorderBuffer().
412  */
413 static void
415 {
417  TransactionId xid = XLogRecGetXid(buf->record);
418  SnapBuild *builder = ctx->snapshot_builder;
419 
420  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
421 
422  /*
423  * If we don't have snapshot or we are just fast-forwarding, there is no
424  * point in decoding data changes.
425  */
427  ctx->fast_forward)
428  return;
429 
430  switch (info)
431  {
432  case XLOG_HEAP_INSERT:
433  if (SnapBuildProcessChange(builder, xid, buf->origptr))
434  DecodeInsert(ctx, buf);
435  break;
436 
437  /*
438  * Treat HOT update as normal updates. There is no useful
439  * information in the fact that we could make it a HOT update
440  * locally and the WAL layout is compatible.
441  */
443  case XLOG_HEAP_UPDATE:
444  if (SnapBuildProcessChange(builder, xid, buf->origptr))
445  DecodeUpdate(ctx, buf);
446  break;
447 
448  case XLOG_HEAP_DELETE:
449  if (SnapBuildProcessChange(builder, xid, buf->origptr))
450  DecodeDelete(ctx, buf);
451  break;
452 
453  case XLOG_HEAP_TRUNCATE:
454  if (SnapBuildProcessChange(builder, xid, buf->origptr))
455  DecodeTruncate(ctx, buf);
456  break;
457 
458  case XLOG_HEAP_INPLACE:
459 
460  /*
461  * Inplace updates are only ever performed on catalog tuples and
462  * can, per definition, not change tuple visibility. Since we
463  * don't decode catalog tuples, we're not interested in the
464  * record's contents.
465  *
466  * In-place updates can be used either by XID-bearing transactions
467  * (e.g. in CREATE INDEX CONCURRENTLY) or by XID-less
468  * transactions (e.g. VACUUM). In the former case, the commit
469  * record will include cache invalidations, so we mark the
470  * transaction as catalog modifying here. Currently that's
471  * redundant because the commit will do that as well, but once we
472  * support decoding in-progress relations, this will be important.
473  */
474  if (!TransactionIdIsValid(xid))
475  break;
476 
477  SnapBuildProcessChange(builder, xid, buf->origptr);
479  break;
480 
481  case XLOG_HEAP_CONFIRM:
482  if (SnapBuildProcessChange(builder, xid, buf->origptr))
483  DecodeSpecConfirm(ctx, buf);
484  break;
485 
486  case XLOG_HEAP_LOCK:
487  /* we don't care about row level locks for now */
488  break;
489 
490  default:
491  elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
492  break;
493  }
494 }
495 
496 static inline bool
498 {
499  if (ctx->callbacks.filter_by_origin_cb == NULL)
500  return false;
501 
502  return filter_by_origin_cb_wrapper(ctx, origin_id);
503 }
504 
505 /*
506  * Handle rmgr LOGICALMSG_ID records for DecodeRecordIntoReorderBuffer().
507  */
508 static void
510 {
511  SnapBuild *builder = ctx->snapshot_builder;
512  XLogReaderState *r = buf->record;
513  TransactionId xid = XLogRecGetXid(r);
514  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
515  RepOriginId origin_id = XLogRecGetOrigin(r);
516  Snapshot snapshot;
517  xl_logical_message *message;
518 
519  if (info != XLOG_LOGICAL_MESSAGE)
520  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
521 
523 
524  /*
525  * If we don't have snapshot or we are just fast-forwarding, there is no
526  * point in decoding messages.
527  */
529  ctx->fast_forward)
530  return;
531 
532  message = (xl_logical_message *) XLogRecGetData(r);
533 
534  if (message->dbId != ctx->slot->data.database ||
535  FilterByOrigin(ctx, origin_id))
536  return;
537 
538  if (message->transactional &&
539  !SnapBuildProcessChange(builder, xid, buf->origptr))
540  return;
541  else if (!message->transactional &&
543  SnapBuildXactNeedsSkip(builder, buf->origptr)))
544  return;
545 
546  snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
547  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
548  message->transactional,
549  message->message, /* first part of message is
550  * prefix */
551  message->message_size,
552  message->message + message->prefix_size);
553 }
554 
555 /*
556  * Consolidated commit record handling between the different form of commit
557  * records.
558  */
559 static void
562 {
563  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
564  TimestampTz commit_time = parsed->xact_time;
565  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
566  int i;
567 
568  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
569  {
570  origin_lsn = parsed->origin_lsn;
571  commit_time = parsed->origin_timestamp;
572  }
573 
574  /*
575  * Process invalidation messages, even if we're not interested in the
576  * transaction's contents, since the various caches need to always be
577  * consistent.
578  */
579  if (parsed->nmsgs > 0)
580  {
581  if (!ctx->fast_forward)
583  parsed->nmsgs, parsed->msgs);
585  }
586 
588  parsed->nsubxacts, parsed->subxacts);
589 
590  /* ----
591  * Check whether we are interested in this specific transaction, and tell
592  * the reorderbuffer to forget the content of the (sub-)transactions
593  * if not.
594  *
595  * There can be several reasons we might not be interested in this
596  * transaction:
597  * 1) We might not be interested in decoding transactions up to this
598  * LSN. This can happen because we previously decoded it and now just
599  * are restarting or if we haven't assembled a consistent snapshot yet.
600  * 2) The transaction happened in another database.
601  * 3) The output plugin is not interested in the origin.
602  * 4) We are doing fast-forwarding
603  *
604  * We can't just use ReorderBufferAbort() here, because we need to execute
605  * the transaction's invalidations. This currently won't be needed if
606  * we're just skipping over the transaction because currently we only do
607  * so during startup, to get to the first transaction the client needs. As
608  * we have reset the catalog caches before starting to read WAL, and we
609  * haven't yet touched any catalogs, there can't be anything to invalidate.
610  * But if we're "forgetting" this commit because it's it happened in
611  * another database, the invalidations might be important, because they
612  * could be for shared catalogs and we might have loaded data into the
613  * relevant syscaches.
614  * ---
615  */
617  (parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
618  ctx->fast_forward || FilterByOrigin(ctx, origin_id))
619  {
620  for (i = 0; i < parsed->nsubxacts; i++)
621  {
622  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
623  }
624  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
625 
626  return;
627  }
628 
629  /* tell the reorderbuffer about the surviving subtransactions */
630  for (i = 0; i < parsed->nsubxacts; i++)
631  {
632  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
633  buf->origptr, buf->endptr);
634  }
635 
636  /* replay actions of all transaction + subtransactions in order */
637  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
638  commit_time, origin_id, origin_lsn);
639 }
640 
641 /*
642  * Get the data from the various forms of abort records and pass it on to
643  * snapbuild.c and reorderbuffer.c
644  */
645 static void
647  xl_xact_parsed_abort *parsed, TransactionId xid)
648 {
649  int i;
650 
651  for (i = 0; i < parsed->nsubxacts; i++)
652  {
653  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
654  buf->record->EndRecPtr);
655  }
656 
657  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
658 }
659 
660 /*
661  * Parse XLOG_HEAP_INSERT (not MULTI_INSERT!) records into tuplebufs.
662  *
663  * Deletes can contain the new tuple.
664  */
665 static void
667 {
668  XLogReaderState *r = buf->record;
669  xl_heap_insert *xlrec;
670  ReorderBufferChange *change;
671  RelFileNode target_node;
672 
673  xlrec = (xl_heap_insert *) XLogRecGetData(r);
674 
675  /* only interested in our database */
676  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
677  if (target_node.dbNode != ctx->slot->data.database)
678  return;
679 
680  /* output plugin doesn't look for this origin, no need to queue */
681  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
682  return;
683 
684  change = ReorderBufferGetChange(ctx->reorder);
685  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
687  else
689  change->origin_id = XLogRecGetOrigin(r);
690 
691  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
692 
694  {
695  Size datalen;
696  char *tupledata = XLogRecGetBlockData(r, 0, &datalen);
697  Size tuplelen = datalen - SizeOfHeapHeader;
698 
699  change->data.tp.newtuple =
700  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
701 
702  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
703  }
704 
705  change->data.tp.clear_toast_afterwards = true;
706 
707  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
708 }
709 
710 /*
711  * Parse XLOG_HEAP_UPDATE and XLOG_HEAP_HOT_UPDATE, which have the same layout
712  * in the record, from wal into proper tuplebufs.
713  *
714  * Updates can possibly contain a new tuple and the old primary key.
715  */
716 static void
718 {
719  XLogReaderState *r = buf->record;
720  xl_heap_update *xlrec;
721  ReorderBufferChange *change;
722  char *data;
723  RelFileNode target_node;
724 
725  xlrec = (xl_heap_update *) XLogRecGetData(r);
726 
727  /* only interested in our database */
728  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
729  if (target_node.dbNode != ctx->slot->data.database)
730  return;
731 
732  /* output plugin doesn't look for this origin, no need to queue */
733  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
734  return;
735 
736  change = ReorderBufferGetChange(ctx->reorder);
738  change->origin_id = XLogRecGetOrigin(r);
739  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
740 
742  {
743  Size datalen;
744  Size tuplelen;
745 
746  data = XLogRecGetBlockData(r, 0, &datalen);
747 
748  tuplelen = datalen - SizeOfHeapHeader;
749 
750  change->data.tp.newtuple =
751  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
752 
753  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
754  }
755 
756  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
757  {
758  Size datalen;
759  Size tuplelen;
760 
761  /* caution, remaining data in record is not aligned */
762  data = XLogRecGetData(r) + SizeOfHeapUpdate;
763  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
764  tuplelen = datalen - SizeOfHeapHeader;
765 
766  change->data.tp.oldtuple =
767  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
768 
769  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
770  }
771 
772  change->data.tp.clear_toast_afterwards = true;
773 
774  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
775 }
776 
777 /*
778  * Parse XLOG_HEAP_DELETE from wal into proper tuplebufs.
779  *
780  * Deletes can possibly contain the old primary key.
781  */
782 static void
784 {
785  XLogReaderState *r = buf->record;
786  xl_heap_delete *xlrec;
787  ReorderBufferChange *change;
788  RelFileNode target_node;
789 
790  xlrec = (xl_heap_delete *) XLogRecGetData(r);
791 
792  /* only interested in our database */
793  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
794  if (target_node.dbNode != ctx->slot->data.database)
795  return;
796 
797  /*
798  * Super deletions are irrelevant for logical decoding, it's driven by the
799  * confirmation records.
800  */
801  if (xlrec->flags & XLH_DELETE_IS_SUPER)
802  return;
803 
804  /* output plugin doesn't look for this origin, no need to queue */
805  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
806  return;
807 
808  change = ReorderBufferGetChange(ctx->reorder);
810  change->origin_id = XLogRecGetOrigin(r);
811 
812  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
813 
814  /* old primary key stored */
815  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
816  {
817  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
818  Size tuplelen = datalen - SizeOfHeapHeader;
819 
820  Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
821 
822  change->data.tp.oldtuple =
823  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
824 
825  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
826  datalen, change->data.tp.oldtuple);
827  }
828 
829  change->data.tp.clear_toast_afterwards = true;
830 
831  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
832 }
833 
834 /*
835  * Parse XLOG_HEAP_TRUNCATE from wal
836  */
837 static void
839 {
840  XLogReaderState *r = buf->record;
841  xl_heap_truncate *xlrec;
842  ReorderBufferChange *change;
843 
844  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
845 
846  /* only interested in our database */
847  if (xlrec->dbId != ctx->slot->data.database)
848  return;
849 
850  /* output plugin doesn't look for this origin, no need to queue */
851  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
852  return;
853 
854  change = ReorderBufferGetChange(ctx->reorder);
856  change->origin_id = XLogRecGetOrigin(r);
857  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
858  change->data.truncate.cascade = true;
859  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
860  change->data.truncate.restart_seqs = true;
861  change->data.truncate.nrelids = xlrec->nrelids;
862  change->data.truncate.relids = palloc(xlrec->nrelids * sizeof(Oid));
863  memcpy(change->data.truncate.relids, xlrec->relids,
864  xlrec->nrelids * sizeof(Oid));
866  buf->origptr, change);
867 }
868 
869 /*
870  * Decode XLOG_HEAP2_MULTI_INSERT_insert record into multiple tuplebufs.
871  *
872  * Currently MULTI_INSERT will always contain the full tuples.
873  */
874 static void
876 {
877  XLogReaderState *r = buf->record;
878  xl_heap_multi_insert *xlrec;
879  int i;
880  char *data;
881  char *tupledata;
882  Size tuplelen;
883  RelFileNode rnode;
884 
885  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
886 
887  /* only interested in our database */
888  XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
889  if (rnode.dbNode != ctx->slot->data.database)
890  return;
891 
892  /* output plugin doesn't look for this origin, no need to queue */
893  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
894  return;
895 
896  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
897 
898  data = tupledata;
899  for (i = 0; i < xlrec->ntuples; i++)
900  {
901  ReorderBufferChange *change;
902  xl_multi_insert_tuple *xlhdr;
903  int datalen;
904  ReorderBufferTupleBuf *tuple;
905 
906  change = ReorderBufferGetChange(ctx->reorder);
908  change->origin_id = XLogRecGetOrigin(r);
909 
910  memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
911 
912  /*
913  * CONTAINS_NEW_TUPLE will always be set currently as multi_insert
914  * isn't used for catalogs, but better be future proof.
915  *
916  * We decode the tuple in pretty much the same way as DecodeXLogTuple,
917  * but since the layout is slightly different, we can't use it here.
918  */
920  {
922 
923  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
924  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
925  datalen = xlhdr->datalen;
926 
927  change->data.tp.newtuple =
928  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
929 
930  tuple = change->data.tp.newtuple;
931  header = tuple->tuple.t_data;
932 
933  /* not a disk based tuple */
935 
936  /*
937  * We can only figure this out after reassembling the
938  * transactions.
939  */
940  tuple->tuple.t_tableOid = InvalidOid;
941 
942  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
943 
944  memset(header, 0, SizeofHeapTupleHeader);
945 
946  memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
947  (char *) data,
948  datalen);
949  data += datalen;
950 
951  header->t_infomask = xlhdr->t_infomask;
952  header->t_infomask2 = xlhdr->t_infomask2;
953  header->t_hoff = xlhdr->t_hoff;
954  }
955 
956  /*
957  * Reset toast reassembly state only after the last row in the last
958  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
959  * call.
960  */
961  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
962  (i + 1) == xlrec->ntuples)
963  change->data.tp.clear_toast_afterwards = true;
964  else
965  change->data.tp.clear_toast_afterwards = false;
966 
968  buf->origptr, change);
969  }
970  Assert(data == tupledata + tuplelen);
971 }
972 
973 /*
974  * Parse XLOG_HEAP_CONFIRM from wal into a confirmation change.
975  *
976  * This is pretty trivial, all the state essentially already setup by the
977  * speculative insertion.
978  */
979 static void
981 {
982  XLogReaderState *r = buf->record;
983  ReorderBufferChange *change;
984  RelFileNode target_node;
985 
986  /* only interested in our database */
987  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
988  if (target_node.dbNode != ctx->slot->data.database)
989  return;
990 
991  /* output plugin doesn't look for this origin, no need to queue */
992  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
993  return;
994 
995  change = ReorderBufferGetChange(ctx->reorder);
997  change->origin_id = XLogRecGetOrigin(r);
998 
999  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1000 
1001  change->data.tp.clear_toast_afterwards = true;
1002 
1003  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
1004 }
1005 
1006 
1007 /*
1008  * Read a HeapTuple as WAL logged by heap_insert, heap_update and heap_delete
1009  * (but not by heap_multi_insert) into a tuplebuf.
1010  *
1011  * The size 'len' and the pointer 'data' in the record need to be
1012  * computed outside as they are record specific.
1013  */
1014 static void
1016 {
1017  xl_heap_header xlhdr;
1018  int datalen = len - SizeOfHeapHeader;
1020 
1021  Assert(datalen >= 0);
1022 
1023  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1024  header = tuple->tuple.t_data;
1025 
1026  /* not a disk based tuple */
1028 
1029  /* we can only figure this out after reassembling the transactions */
1030  tuple->tuple.t_tableOid = InvalidOid;
1031 
1032  /* data is not stored aligned, copy to aligned storage */
1033  memcpy((char *) &xlhdr,
1034  data,
1035  SizeOfHeapHeader);
1036 
1037  memset(header, 0, SizeofHeapTupleHeader);
1038 
1039  memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
1040  data + SizeOfHeapHeader,
1041  datalen);
1042 
1043  header->t_infomask = xlhdr.t_infomask;
1044  header->t_infomask2 = xlhdr.t_infomask2;
1045  header->t_hoff = xlhdr.t_hoff;
1046 }
struct ReorderBufferChange::@102::@104 truncate
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:875
#define XLH_TRUNCATE_CASCADE
Definition: heapam_xlog.h:116
#define SizeofHeapTupleHeader
Definition: htup_details.h:181
#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:140
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid)
Definition: decode.c:560
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:164
#define XLOG_FPW_CHANGE
Definition: pg_control.h:75
struct ReorderBuffer * reorder
Definition: logical.h:45
#define XLOG_XACT_PREPARE
Definition: xact.h:141
static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:666
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:59
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
uint32 TransactionId
Definition: c.h:474
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:410
#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:320
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:927
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:401
int64 TimestampTz
Definition: timestamp.h:39
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:53
TransactionId twophase_xid
Definition: xact.h:314
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1015
#define XLOG_INVALIDATIONS
Definition: standbydefs.h:36
OutputPluginCallbacks callbacks
Definition: logical.h:55
unsigned char uint8
Definition: c.h:323
#define XLOG_STANDBY_LOCK
Definition: standbydefs.h:34
TransactionId * subxacts
Definition: xact.h:306
#define XLH_INSERT_IS_SPECULATIVE
Definition: heapam_xlog.h:68
uint16 RepOriginId
Definition: xlogdefs.h:51
#define XLOG_RESTORE_POINT
Definition: pg_control.h:74
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
SharedInvalidationMessage * msgs
Definition: xact.h:312
TransactionId xtop
Definition: xact.h:190
XLogRecPtr origin_lsn
Definition: xact.h:319
#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:192
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:120
unsigned int Oid
Definition: postgres_ext.h:31
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
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:717
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid)
Definition: decode.c:646
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:225
#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:120
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:719
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
struct ReorderBufferChange::@102::@103 tp
#define XLOG_NEXTOID
Definition: pg_control.h:70
static void DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:509
#define XLogRecGetData(decoder)
Definition: xlogreader.h:226
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:980
#define ERROR
Definition: elog.h:43
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:97
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:227
#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:783
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:67
#define XLOG_HEAP2_CLEANUP_INFO
Definition: heapam_xlog.h:56
#define XLOG_XACT_ABORT_PREPARED
Definition: xact.h:144
#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:769
XLogRecPtr ReadRecPtr
Definition: xlogreader.h:119
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:1094
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:60
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:781
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:222
RepOriginId origin_id
Definition: reorderbuffer.h:80
bool transactional
Definition: message.h:23
TransactionId * subxacts
Definition: xact.h:334
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:1330
static void DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:354
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:224
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:145
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1354
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
#define XLOG_NOOP
Definition: pg_control.h:69
union ReorderBufferChange::@102 data
#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:339
struct SnapBuild * snapshot_builder
Definition: logical.h:46
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:699
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
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:142
#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:433
#define XLOG_XACT_OPMASK
Definition: xact.h:150
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:838
#define XLOG_HEAP_INPLACE
Definition: heapam_xlog.h:39
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:208
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:150
#define XLOG_FPI
Definition: pg_control.h:78
void * palloc(Size size)
Definition: mcxt.c:924
XLogReaderState * reader
Definition: logical.h:44
ReplicationSlot * slot
Definition: logical.h:41
#define XLOG_HEAP_LOCK
Definition: heapam_xlog.h:38
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
int i
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1464
#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:414
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define elog
Definition: elog.h:219
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define XLH_UPDATE_CONTAINS_OLD
Definition: heapam_xlog.h:85
#define SHORTALIGN(LEN)
Definition: c.h:648
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:143
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:304
#define SizeOfHeapHeader
Definition: heapam_xlog.h:149
#define XLogRecGetRmid(decoder)
Definition: xlogreader.h:223
#define XLOG_HEAP_CONFIRM
Definition: heapam_xlog.h:37
TimestampTz xact_time
Definition: xact.h:299
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:191
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:675
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)