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