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-2021, 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/xlogreader.h"
35 #include "access/xlogrecord.h"
36 #include "access/xlogutils.h"
37 #include "catalog/pg_control.h"
38 #include "replication/decode.h"
39 #include "replication/logical.h"
40 #include "replication/message.h"
41 #include "replication/origin.h"
43 #include "replication/snapbuild.h"
44 #include "storage/standby.h"
45 
46 typedef struct XLogRecordBuffer
47 {
52 
53 /* RMGR Handlers */
60 
61 /* individual record(group)'s handlers */
68 
71  bool two_phase);
74  bool two_phase);
76  xl_xact_parsed_prepare *parsed);
77 
78 
79 /* common function to decode tuples */
80 static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup);
81 
82 /* helper functions for decoding transactions */
83 static inline bool FilterPrepare(LogicalDecodingContext *ctx,
84  TransactionId xid, const char *gid);
86  XLogRecordBuffer *buf, Oid dbId,
87  RepOriginId origin_id);
88 
89 /*
90  * Take every XLogReadRecord()ed record and perform the actions required to
91  * decode it using the output plugin already setup in the logical decoding
92  * context.
93  *
94  * NB: Note that every record's xid needs to be processed by reorderbuffer
95  * (xids contained in the content of records are not relevant for this rule).
96  * That means that for records which'd otherwise not go through the
97  * reorderbuffer ReorderBufferProcessXid() has to be called. We don't want to
98  * call ReorderBufferProcessXid for each record type by default, because
99  * e.g. empty xacts can be handled more efficiently if there's no previous
100  * state for them.
101  *
102  * We also support the ability to fast forward thru records, skipping some
103  * record types completely - see individual record types for details.
104  */
105 void
107 {
109  TransactionId txid;
110 
111  buf.origptr = ctx->reader->ReadRecPtr;
112  buf.endptr = ctx->reader->EndRecPtr;
113  buf.record = record;
114 
115  txid = XLogRecGetTopXid(record);
116 
117  /*
118  * If the top-level xid is valid, we need to assign the subxact to the
119  * top-level xact. We need to do this for all records, hence we do it
120  * before the switch.
121  */
122  if (TransactionIdIsValid(txid))
123  {
125  txid,
126  XLogRecGetXid(record),
127  buf.origptr);
128  }
129 
130  /* cast so we get a warning when new rmgrs are added */
131  switch ((RmgrId) XLogRecGetRmid(record))
132  {
133  /*
134  * Rmgrs we care about for logical decoding. Add new rmgrs in
135  * rmgrlist.h's order.
136  */
137  case RM_XLOG_ID:
138  DecodeXLogOp(ctx, &buf);
139  break;
140 
141  case RM_XACT_ID:
142  DecodeXactOp(ctx, &buf);
143  break;
144 
145  case RM_STANDBY_ID:
146  DecodeStandbyOp(ctx, &buf);
147  break;
148 
149  case RM_HEAP2_ID:
150  DecodeHeap2Op(ctx, &buf);
151  break;
152 
153  case RM_HEAP_ID:
154  DecodeHeapOp(ctx, &buf);
155  break;
156 
157  case RM_LOGICALMSG_ID:
158  DecodeLogicalMsgOp(ctx, &buf);
159  break;
160 
161  /*
162  * Rmgrs irrelevant for logical decoding; they describe stuff not
163  * represented in logical decoding. Add new rmgrs in rmgrlist.h's
164  * order.
165  */
166  case RM_SMGR_ID:
167  case RM_CLOG_ID:
168  case RM_DBASE_ID:
169  case RM_TBLSPC_ID:
170  case RM_MULTIXACT_ID:
171  case RM_RELMAP_ID:
172  case RM_BTREE_ID:
173  case RM_HASH_ID:
174  case RM_GIN_ID:
175  case RM_GIST_ID:
176  case RM_SEQ_ID:
177  case RM_SPGIST_ID:
178  case RM_BRIN_ID:
179  case RM_COMMIT_TS_ID:
180  case RM_REPLORIGIN_ID:
181  case RM_GENERIC_ID:
182  /* just deal with xid, and done */
184  buf.origptr);
185  break;
186  case RM_NEXT_ID:
187  elog(ERROR, "unexpected RM_NEXT_ID rmgr_id: %u", (RmgrIds) XLogRecGetRmid(buf.record));
188  }
189 }
190 
191 /*
192  * Handle rmgr XLOG_ID records for DecodeRecordIntoReorderBuffer().
193  */
194 static void
196 {
197  SnapBuild *builder = ctx->snapshot_builder;
198  uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
199 
201  buf->origptr);
202 
203  switch (info)
204  {
205  /* this is also used in END_OF_RECOVERY checkpoints */
208  SnapBuildSerializationPoint(builder, buf->origptr);
209 
210  break;
212 
213  /*
214  * a RUNNING_XACTS record will have been logged near to this, we
215  * can restart from there.
216  */
217  break;
218  case XLOG_NOOP:
219  case XLOG_NEXTOID:
220  case XLOG_SWITCH:
221  case XLOG_BACKUP_END:
223  case XLOG_RESTORE_POINT:
224  case XLOG_FPW_CHANGE:
225  case XLOG_FPI_FOR_HINT:
226  case XLOG_FPI:
227  break;
228  default:
229  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
230  }
231 }
232 
233 /*
234  * Handle rmgr XACT_ID records for DecodeRecordIntoReorderBuffer().
235  */
236 static void
238 {
239  SnapBuild *builder = ctx->snapshot_builder;
240  ReorderBuffer *reorder = ctx->reorder;
241  XLogReaderState *r = buf->record;
243 
244  /*
245  * If the snapshot isn't yet fully built, we cannot decode anything, so
246  * bail out.
247  */
249  return;
250 
251  switch (info)
252  {
253  case XLOG_XACT_COMMIT:
255  {
256  xl_xact_commit *xlrec;
257  xl_xact_parsed_commit parsed;
258  TransactionId xid;
259  bool two_phase = false;
260 
261  xlrec = (xl_xact_commit *) XLogRecGetData(r);
262  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
263 
264  if (!TransactionIdIsValid(parsed.twophase_xid))
265  xid = XLogRecGetXid(r);
266  else
267  xid = parsed.twophase_xid;
268 
269  /*
270  * We would like to process the transaction in a two-phase
271  * manner iff output plugin supports two-phase commits and
272  * doesn't filter the transaction at prepare time.
273  */
274  if (info == XLOG_XACT_COMMIT_PREPARED)
275  two_phase = !(FilterPrepare(ctx, xid,
276  parsed.twophase_gid));
277 
278  DecodeCommit(ctx, buf, &parsed, xid, two_phase);
279  break;
280  }
281  case XLOG_XACT_ABORT:
283  {
284  xl_xact_abort *xlrec;
285  xl_xact_parsed_abort parsed;
286  TransactionId xid;
287  bool two_phase = false;
288 
289  xlrec = (xl_xact_abort *) XLogRecGetData(r);
290  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
291 
292  if (!TransactionIdIsValid(parsed.twophase_xid))
293  xid = XLogRecGetXid(r);
294  else
295  xid = parsed.twophase_xid;
296 
297  /*
298  * We would like to process the transaction in a two-phase
299  * manner iff output plugin supports two-phase commits and
300  * doesn't filter the transaction at prepare time.
301  */
302  if (info == XLOG_XACT_ABORT_PREPARED)
303  two_phase = !(FilterPrepare(ctx, xid,
304  parsed.twophase_gid));
305 
306  DecodeAbort(ctx, buf, &parsed, xid, two_phase);
307  break;
308  }
310 
311  /*
312  * We assign subxact to the toplevel xact while processing each
313  * record if required. So, we don't need to do anything here. See
314  * LogicalDecodingProcessRecord.
315  */
316  break;
318  {
319  TransactionId xid;
320  xl_xact_invals *invals;
321 
322  xid = XLogRecGetXid(r);
323  invals = (xl_xact_invals *) XLogRecGetData(r);
324 
325  /*
326  * Execute the invalidations for xid-less transactions,
327  * otherwise, accumulate them so that they can be processed at
328  * the commit time.
329  */
330  if (TransactionIdIsValid(xid))
331  {
332  if (!ctx->fast_forward)
333  ReorderBufferAddInvalidations(reorder, xid,
334  buf->origptr,
335  invals->nmsgs,
336  invals->msgs);
338  buf->origptr);
339  }
340  else if ((!ctx->fast_forward))
342  invals->nmsgs,
343  invals->msgs);
344  }
345  break;
346  case XLOG_XACT_PREPARE:
347  {
348  xl_xact_parsed_prepare parsed;
349  xl_xact_prepare *xlrec;
350 
351  /* ok, parse it */
352  xlrec = (xl_xact_prepare *) XLogRecGetData(r);
354  xlrec, &parsed);
355 
356  /*
357  * We would like to process the transaction in a two-phase
358  * manner iff output plugin supports two-phase commits and
359  * doesn't filter the transaction at prepare time.
360  */
361  if (FilterPrepare(ctx, parsed.twophase_xid,
362  parsed.twophase_gid))
363  {
364  ReorderBufferProcessXid(reorder, parsed.twophase_xid,
365  buf->origptr);
366  break;
367  }
368 
369  /*
370  * Note that if the prepared transaction has locked [user]
371  * catalog tables exclusively then decoding prepare can block
372  * till the main transaction is committed because it needs to
373  * lock the catalog tables.
374  *
375  * XXX Now, this can even lead to a deadlock if the prepare
376  * transaction is waiting to get it logically replicated for
377  * distributed 2PC. Currently, we don't have an in-core
378  * implementation of prepares for distributed 2PC but some
379  * out-of-core logical replication solution can have such an
380  * implementation. They need to inform users to not have locks
381  * on catalog tables in such transactions.
382  */
383  DecodePrepare(ctx, buf, &parsed);
384  break;
385  }
386  default:
387  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
388  }
389 }
390 
391 /*
392  * Handle rmgr STANDBY_ID records for DecodeRecordIntoReorderBuffer().
393  */
394 static void
396 {
397  SnapBuild *builder = ctx->snapshot_builder;
398  XLogReaderState *r = buf->record;
399  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
400 
402 
403  switch (info)
404  {
405  case XLOG_RUNNING_XACTS:
406  {
408 
409  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
410 
411  /*
412  * Abort all transactions that we keep track of, that are
413  * older than the record's oldestRunningXid. This is the most
414  * convenient spot for doing so since, in contrast to shutdown
415  * or end-of-recovery checkpoints, we have information about
416  * all running transactions which includes prepared ones,
417  * while shutdown checkpoints just know that no non-prepared
418  * transactions are in progress.
419  */
421  }
422  break;
423  case XLOG_STANDBY_LOCK:
424  break;
425  case XLOG_INVALIDATIONS:
426 
427  /*
428  * We are processing the invalidations at the command level via
429  * XLOG_XACT_INVALIDATIONS. So we don't need to do anything here.
430  */
431  break;
432  default:
433  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
434  }
435 }
436 
437 /*
438  * Handle rmgr HEAP2_ID records for DecodeRecordIntoReorderBuffer().
439  */
440 static void
442 {
444  TransactionId xid = XLogRecGetXid(buf->record);
445  SnapBuild *builder = ctx->snapshot_builder;
446 
447  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
448 
449  /*
450  * If we don't have snapshot or we are just fast-forwarding, there is no
451  * point in decoding changes.
452  */
454  ctx->fast_forward)
455  return;
456 
457  switch (info)
458  {
460  if (!ctx->fast_forward &&
461  SnapBuildProcessChange(builder, xid, buf->origptr))
462  DecodeMultiInsert(ctx, buf);
463  break;
464  case XLOG_HEAP2_NEW_CID:
465  {
466  xl_heap_new_cid *xlrec;
467 
468  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
469  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
470 
471  break;
472  }
473  case XLOG_HEAP2_REWRITE:
474 
475  /*
476  * Although these records only exist to serve the needs of logical
477  * decoding, all the work happens as part of crash or archive
478  * recovery, so we don't need to do anything here.
479  */
480  break;
481 
482  /*
483  * Everything else here is just low level physical stuff we're not
484  * interested in.
485  */
487  case XLOG_HEAP2_PRUNE:
488  case XLOG_HEAP2_VACUUM:
489  case XLOG_HEAP2_VISIBLE:
491  break;
492  default:
493  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
494  }
495 }
496 
497 /*
498  * Handle rmgr HEAP_ID records for DecodeRecordIntoReorderBuffer().
499  */
500 static void
502 {
504  TransactionId xid = XLogRecGetXid(buf->record);
505  SnapBuild *builder = ctx->snapshot_builder;
506 
507  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
508 
509  /*
510  * If we don't have snapshot or we are just fast-forwarding, there is no
511  * point in decoding data changes.
512  */
514  ctx->fast_forward)
515  return;
516 
517  switch (info)
518  {
519  case XLOG_HEAP_INSERT:
520  if (SnapBuildProcessChange(builder, xid, buf->origptr))
521  DecodeInsert(ctx, buf);
522  break;
523 
524  /*
525  * Treat HOT update as normal updates. There is no useful
526  * information in the fact that we could make it a HOT update
527  * locally and the WAL layout is compatible.
528  */
530  case XLOG_HEAP_UPDATE:
531  if (SnapBuildProcessChange(builder, xid, buf->origptr))
532  DecodeUpdate(ctx, buf);
533  break;
534 
535  case XLOG_HEAP_DELETE:
536  if (SnapBuildProcessChange(builder, xid, buf->origptr))
537  DecodeDelete(ctx, buf);
538  break;
539 
540  case XLOG_HEAP_TRUNCATE:
541  if (SnapBuildProcessChange(builder, xid, buf->origptr))
542  DecodeTruncate(ctx, buf);
543  break;
544 
545  case XLOG_HEAP_INPLACE:
546 
547  /*
548  * Inplace updates are only ever performed on catalog tuples and
549  * can, per definition, not change tuple visibility. Since we
550  * don't decode catalog tuples, we're not interested in the
551  * record's contents.
552  *
553  * In-place updates can be used either by XID-bearing transactions
554  * (e.g. in CREATE INDEX CONCURRENTLY) or by XID-less
555  * transactions (e.g. VACUUM). In the former case, the commit
556  * record will include cache invalidations, so we mark the
557  * transaction as catalog modifying here. Currently that's
558  * redundant because the commit will do that as well, but once we
559  * support decoding in-progress relations, this will be important.
560  */
561  if (!TransactionIdIsValid(xid))
562  break;
563 
564  SnapBuildProcessChange(builder, xid, buf->origptr);
566  break;
567 
568  case XLOG_HEAP_CONFIRM:
569  if (SnapBuildProcessChange(builder, xid, buf->origptr))
570  DecodeSpecConfirm(ctx, buf);
571  break;
572 
573  case XLOG_HEAP_LOCK:
574  /* we don't care about row level locks for now */
575  break;
576 
577  default:
578  elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
579  break;
580  }
581 }
582 
583 /*
584  * Ask output plugin whether we want to skip this PREPARE and send
585  * this transaction as a regular commit later.
586  */
587 static inline bool
589  const char *gid)
590 {
591  /*
592  * Skip if decoding of two-phase transactions at PREPARE time is not
593  * enabled. In that case, all two-phase transactions are considered
594  * filtered out and will be applied as regular transactions at COMMIT
595  * PREPARED.
596  */
597  if (!ctx->twophase)
598  return true;
599 
600  /*
601  * The filter_prepare callback is optional. When not supplied, all
602  * prepared transactions should go through.
603  */
604  if (ctx->callbacks.filter_prepare_cb == NULL)
605  return false;
606 
607  return filter_prepare_cb_wrapper(ctx, xid, gid);
608 }
609 
610 static inline bool
612 {
613  if (ctx->callbacks.filter_by_origin_cb == NULL)
614  return false;
615 
616  return filter_by_origin_cb_wrapper(ctx, origin_id);
617 }
618 
619 /*
620  * Handle rmgr LOGICALMSG_ID records for DecodeRecordIntoReorderBuffer().
621  */
622 static void
624 {
625  SnapBuild *builder = ctx->snapshot_builder;
626  XLogReaderState *r = buf->record;
627  TransactionId xid = XLogRecGetXid(r);
628  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
629  RepOriginId origin_id = XLogRecGetOrigin(r);
630  Snapshot snapshot;
631  xl_logical_message *message;
632 
633  if (info != XLOG_LOGICAL_MESSAGE)
634  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
635 
637 
638  /*
639  * If we don't have snapshot or we are just fast-forwarding, there is no
640  * point in decoding messages.
641  */
643  ctx->fast_forward)
644  return;
645 
646  message = (xl_logical_message *) XLogRecGetData(r);
647 
648  if (message->dbId != ctx->slot->data.database ||
649  FilterByOrigin(ctx, origin_id))
650  return;
651 
652  if (message->transactional &&
653  !SnapBuildProcessChange(builder, xid, buf->origptr))
654  return;
655  else if (!message->transactional &&
657  SnapBuildXactNeedsSkip(builder, buf->origptr)))
658  return;
659 
660  snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
661  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
662  message->transactional,
663  message->message, /* first part of message is
664  * prefix */
665  message->message_size,
666  message->message + message->prefix_size);
667 }
668 
669 /*
670  * Consolidated commit record handling between the different form of commit
671  * records.
672  *
673  * 'two_phase' indicates that caller wants to process the transaction in two
674  * phases, first process prepare if not already done and then process
675  * commit_prepared.
676  */
677 static void
680  bool two_phase)
681 {
682  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
683  TimestampTz commit_time = parsed->xact_time;
684  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
685  int i;
686 
687  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
688  {
689  origin_lsn = parsed->origin_lsn;
690  commit_time = parsed->origin_timestamp;
691  }
692 
694  parsed->nsubxacts, parsed->subxacts);
695 
696  /* ----
697  * Check whether we are interested in this specific transaction, and tell
698  * the reorderbuffer to forget the content of the (sub-)transactions
699  * if not.
700  *
701  * We can't just use ReorderBufferAbort() here, because we need to execute
702  * the transaction's invalidations. This currently won't be needed if
703  * we're just skipping over the transaction because currently we only do
704  * so during startup, to get to the first transaction the client needs. As
705  * we have reset the catalog caches before starting to read WAL, and we
706  * haven't yet touched any catalogs, there can't be anything to invalidate.
707  * But if we're "forgetting" this commit because it happened in another
708  * database, the invalidations might be important, because they could be
709  * for shared catalogs and we might have loaded data into the relevant
710  * syscaches.
711  * ---
712  */
713  if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
714  {
715  for (i = 0; i < parsed->nsubxacts; i++)
716  {
717  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
718  }
719  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
720 
721  return;
722  }
723 
724  /* tell the reorderbuffer about the surviving subtransactions */
725  for (i = 0; i < parsed->nsubxacts; i++)
726  {
727  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
728  buf->origptr, buf->endptr);
729  }
730 
731  /*
732  * Send the final commit record if the transaction data is already
733  * decoded, otherwise, process the entire transaction.
734  */
735  if (two_phase)
736  {
737  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
739  commit_time, origin_id, origin_lsn,
740  parsed->twophase_gid, true);
741  }
742  else
743  {
744  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
745  commit_time, origin_id, origin_lsn);
746  }
747 
748  /*
749  * Update the decoding stats at transaction prepare/commit/abort. It is
750  * not clear that sending more or less frequently than this would be
751  * better.
752  */
753  UpdateDecodingStats(ctx);
754 }
755 
756 /*
757  * Decode PREPARE record. Similar logic as in DecodeCommit.
758  *
759  * Note that we don't skip prepare even if have detected concurrent abort
760  * because it is quite possible that we had already sent some changes before we
761  * detect abort in which case we need to abort those changes in the subscriber.
762  * To abort such changes, we do send the prepare and then the rollback prepared
763  * which is what happened on the publisher-side as well. Now, we can invent a
764  * new abort API wherein in such cases we send abort and skip sending prepared
765  * and rollback prepared but then it is not that straightforward because we
766  * might have streamed this transaction by that time in which case it is
767  * handled when the rollback is encountered. It is not impossible to optimize
768  * the concurrent abort case but it can introduce design complexity w.r.t
769  * handling different cases so leaving it for now as it doesn't seem worth it.
770  */
771 static void
773  xl_xact_parsed_prepare *parsed)
774 {
775  SnapBuild *builder = ctx->snapshot_builder;
776  XLogRecPtr origin_lsn = parsed->origin_lsn;
777  TimestampTz prepare_time = parsed->xact_time;
778  XLogRecPtr origin_id = XLogRecGetOrigin(buf->record);
779  int i;
780  TransactionId xid = parsed->twophase_xid;
781 
782  if (parsed->origin_timestamp != 0)
783  prepare_time = parsed->origin_timestamp;
784 
785  /*
786  * Remember the prepare info for a txn so that it can be used later in
787  * commit prepared if required. See ReorderBufferFinishPrepared.
788  */
789  if (!ReorderBufferRememberPrepareInfo(ctx->reorder, xid, buf->origptr,
790  buf->endptr, prepare_time, origin_id,
791  origin_lsn))
792  return;
793 
794  /* We can't start streaming unless a consistent state is reached. */
796  {
798  return;
799  }
800 
801  /*
802  * Check whether we need to process this transaction. See
803  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
804  * transaction.
805  *
806  * We can't call ReorderBufferForget as we did in DecodeCommit as the txn
807  * hasn't yet been committed, removing this txn before a commit might
808  * result in the computation of an incorrect restart_lsn. See
809  * SnapBuildProcessRunningXacts. But we need to process cache
810  * invalidations if there are any for the reasons mentioned in
811  * DecodeCommit.
812  */
813  if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
814  {
816  ReorderBufferInvalidate(ctx->reorder, xid, buf->origptr);
817  return;
818  }
819 
820  /* Tell the reorderbuffer about the surviving subtransactions. */
821  for (i = 0; i < parsed->nsubxacts; i++)
822  {
823  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
824  buf->origptr, buf->endptr);
825  }
826 
827  /* replay actions of all transaction + subtransactions in order */
828  ReorderBufferPrepare(ctx->reorder, xid, parsed->twophase_gid);
829 
830  /*
831  * Update the decoding stats at transaction prepare/commit/abort. It is
832  * not clear that sending more or less frequently than this would be
833  * better.
834  */
835  UpdateDecodingStats(ctx);
836 }
837 
838 
839 /*
840  * Get the data from the various forms of abort records and pass it on to
841  * snapbuild.c and reorderbuffer.c.
842  *
843  * 'two_phase' indicates to finish prepared transaction.
844  */
845 static void
847  xl_xact_parsed_abort *parsed, TransactionId xid,
848  bool two_phase)
849 {
850  int i;
851  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
852  TimestampTz abort_time = parsed->xact_time;
853  XLogRecPtr origin_id = XLogRecGetOrigin(buf->record);
854  bool skip_xact;
855 
856  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
857  {
858  origin_lsn = parsed->origin_lsn;
859  abort_time = parsed->origin_timestamp;
860  }
861 
862  /*
863  * Check whether we need to process this transaction. See
864  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
865  * transaction.
866  */
867  skip_xact = DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id);
868 
869  /*
870  * Send the final rollback record for a prepared transaction unless we
871  * need to skip it. For non-two-phase xacts, simply forget the xact.
872  */
873  if (two_phase && !skip_xact)
874  {
875  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
876  abort_time, origin_id, origin_lsn,
878  parsed->twophase_gid, false);
879  }
880  else
881  {
882  for (i = 0; i < parsed->nsubxacts; i++)
883  {
884  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
885  buf->record->EndRecPtr);
886  }
887 
888  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
889  }
890 
891  /* update the decoding stats */
892  UpdateDecodingStats(ctx);
893 }
894 
895 /*
896  * Parse XLOG_HEAP_INSERT (not MULTI_INSERT!) records into tuplebufs.
897  *
898  * Deletes can contain the new tuple.
899  */
900 static void
902 {
903  Size datalen;
904  char *tupledata;
905  Size tuplelen;
906  XLogReaderState *r = buf->record;
907  xl_heap_insert *xlrec;
908  ReorderBufferChange *change;
909  RelFileNode target_node;
910 
911  xlrec = (xl_heap_insert *) XLogRecGetData(r);
912 
913  /*
914  * Ignore insert records without new tuples (this does happen when
915  * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
916  */
917  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
918  return;
919 
920  /* only interested in our database */
921  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
922  if (target_node.dbNode != ctx->slot->data.database)
923  return;
924 
925  /* output plugin doesn't look for this origin, no need to queue */
926  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
927  return;
928 
929  change = ReorderBufferGetChange(ctx->reorder);
930  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
932  else
934  change->origin_id = XLogRecGetOrigin(r);
935 
936  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
937 
938  tupledata = XLogRecGetBlockData(r, 0, &datalen);
939  tuplelen = datalen - SizeOfHeapHeader;
940 
941  change->data.tp.newtuple =
942  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
943 
944  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
945 
946  change->data.tp.clear_toast_afterwards = true;
947 
949  change,
951 }
952 
953 /*
954  * Parse XLOG_HEAP_UPDATE and XLOG_HEAP_HOT_UPDATE, which have the same layout
955  * in the record, from wal into proper tuplebufs.
956  *
957  * Updates can possibly contain a new tuple and the old primary key.
958  */
959 static void
961 {
962  XLogReaderState *r = buf->record;
963  xl_heap_update *xlrec;
964  ReorderBufferChange *change;
965  char *data;
966  RelFileNode target_node;
967 
968  xlrec = (xl_heap_update *) XLogRecGetData(r);
969 
970  /* only interested in our database */
971  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
972  if (target_node.dbNode != ctx->slot->data.database)
973  return;
974 
975  /* output plugin doesn't look for this origin, no need to queue */
976  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
977  return;
978 
979  change = ReorderBufferGetChange(ctx->reorder);
981  change->origin_id = XLogRecGetOrigin(r);
982  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
983 
985  {
986  Size datalen;
987  Size tuplelen;
988 
989  data = XLogRecGetBlockData(r, 0, &datalen);
990 
991  tuplelen = datalen - SizeOfHeapHeader;
992 
993  change->data.tp.newtuple =
994  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
995 
996  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
997  }
998 
999  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
1000  {
1001  Size datalen;
1002  Size tuplelen;
1003 
1004  /* caution, remaining data in record is not aligned */
1005  data = XLogRecGetData(r) + SizeOfHeapUpdate;
1006  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
1007  tuplelen = datalen - SizeOfHeapHeader;
1008 
1009  change->data.tp.oldtuple =
1010  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1011 
1012  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
1013  }
1014 
1015  change->data.tp.clear_toast_afterwards = true;
1016 
1018  change, false);
1019 }
1020 
1021 /*
1022  * Parse XLOG_HEAP_DELETE from wal into proper tuplebufs.
1023  *
1024  * Deletes can possibly contain the old primary key.
1025  */
1026 static void
1028 {
1029  XLogReaderState *r = buf->record;
1030  xl_heap_delete *xlrec;
1031  ReorderBufferChange *change;
1032  RelFileNode target_node;
1033 
1034  xlrec = (xl_heap_delete *) XLogRecGetData(r);
1035 
1036  /* only interested in our database */
1037  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
1038  if (target_node.dbNode != ctx->slot->data.database)
1039  return;
1040 
1041  /*
1042  * Super deletions are irrelevant for logical decoding, it's driven by the
1043  * confirmation records.
1044  */
1045  if (xlrec->flags & XLH_DELETE_IS_SUPER)
1046  return;
1047 
1048  /* output plugin doesn't look for this origin, no need to queue */
1049  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1050  return;
1051 
1052  change = ReorderBufferGetChange(ctx->reorder);
1054  change->origin_id = XLogRecGetOrigin(r);
1055 
1056  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1057 
1058  /* old primary key stored */
1059  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
1060  {
1061  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
1062  Size tuplelen = datalen - SizeOfHeapHeader;
1063 
1064  Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
1065 
1066  change->data.tp.oldtuple =
1067  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1068 
1069  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
1070  datalen, change->data.tp.oldtuple);
1071  }
1072 
1073  change->data.tp.clear_toast_afterwards = true;
1074 
1076  change, false);
1077 }
1078 
1079 /*
1080  * Parse XLOG_HEAP_TRUNCATE from wal
1081  */
1082 static void
1084 {
1085  XLogReaderState *r = buf->record;
1086  xl_heap_truncate *xlrec;
1087  ReorderBufferChange *change;
1088 
1089  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
1090 
1091  /* only interested in our database */
1092  if (xlrec->dbId != ctx->slot->data.database)
1093  return;
1094 
1095  /* output plugin doesn't look for this origin, no need to queue */
1096  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1097  return;
1098 
1099  change = ReorderBufferGetChange(ctx->reorder);
1101  change->origin_id = XLogRecGetOrigin(r);
1102  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
1103  change->data.truncate.cascade = true;
1104  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
1105  change->data.truncate.restart_seqs = true;
1106  change->data.truncate.nrelids = xlrec->nrelids;
1107  change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
1108  xlrec->nrelids);
1109  memcpy(change->data.truncate.relids, xlrec->relids,
1110  xlrec->nrelids * sizeof(Oid));
1112  buf->origptr, change, false);
1113 }
1114 
1115 /*
1116  * Decode XLOG_HEAP2_MULTI_INSERT_insert record into multiple tuplebufs.
1117  *
1118  * Currently MULTI_INSERT will always contain the full tuples.
1119  */
1120 static void
1122 {
1123  XLogReaderState *r = buf->record;
1124  xl_heap_multi_insert *xlrec;
1125  int i;
1126  char *data;
1127  char *tupledata;
1128  Size tuplelen;
1129  RelFileNode rnode;
1130 
1131  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
1132 
1133  /*
1134  * Ignore insert records without new tuples. This happens when a
1135  * multi_insert is done on a catalog or on a non-persistent relation.
1136  */
1137  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
1138  return;
1139 
1140  /* only interested in our database */
1141  XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
1142  if (rnode.dbNode != ctx->slot->data.database)
1143  return;
1144 
1145  /* output plugin doesn't look for this origin, no need to queue */
1146  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1147  return;
1148 
1149  /*
1150  * We know that this multi_insert isn't for a catalog, so the block should
1151  * always have data even if a full-page write of it is taken.
1152  */
1153  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
1154  Assert(tupledata != NULL);
1155 
1156  data = tupledata;
1157  for (i = 0; i < xlrec->ntuples; i++)
1158  {
1159  ReorderBufferChange *change;
1160  xl_multi_insert_tuple *xlhdr;
1161  int datalen;
1162  ReorderBufferTupleBuf *tuple;
1164 
1165  change = ReorderBufferGetChange(ctx->reorder);
1167  change->origin_id = XLogRecGetOrigin(r);
1168 
1169  memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
1170 
1171  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
1172  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
1173  datalen = xlhdr->datalen;
1174 
1175  change->data.tp.newtuple =
1176  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
1177 
1178  tuple = change->data.tp.newtuple;
1179  header = tuple->tuple.t_data;
1180 
1181  /* not a disk based tuple */
1183 
1184  /*
1185  * We can only figure this out after reassembling the transactions.
1186  */
1187  tuple->tuple.t_tableOid = InvalidOid;
1188 
1189  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1190 
1191  memset(header, 0, SizeofHeapTupleHeader);
1192 
1193  memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
1194  (char *) data,
1195  datalen);
1196  header->t_infomask = xlhdr->t_infomask;
1197  header->t_infomask2 = xlhdr->t_infomask2;
1198  header->t_hoff = xlhdr->t_hoff;
1199 
1200  /*
1201  * Reset toast reassembly state only after the last row in the last
1202  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
1203  * call.
1204  */
1205  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
1206  (i + 1) == xlrec->ntuples)
1207  change->data.tp.clear_toast_afterwards = true;
1208  else
1209  change->data.tp.clear_toast_afterwards = false;
1210 
1212  buf->origptr, change, false);
1213 
1214  /* move to the next xl_multi_insert_tuple entry */
1215  data += datalen;
1216  }
1217  Assert(data == tupledata + tuplelen);
1218 }
1219 
1220 /*
1221  * Parse XLOG_HEAP_CONFIRM from wal into a confirmation change.
1222  *
1223  * This is pretty trivial, all the state essentially already setup by the
1224  * speculative insertion.
1225  */
1226 static void
1228 {
1229  XLogReaderState *r = buf->record;
1230  ReorderBufferChange *change;
1231  RelFileNode target_node;
1232 
1233  /* only interested in our database */
1234  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
1235  if (target_node.dbNode != ctx->slot->data.database)
1236  return;
1237 
1238  /* output plugin doesn't look for this origin, no need to queue */
1239  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1240  return;
1241 
1242  change = ReorderBufferGetChange(ctx->reorder);
1244  change->origin_id = XLogRecGetOrigin(r);
1245 
1246  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1247 
1248  change->data.tp.clear_toast_afterwards = true;
1249 
1251  change, false);
1252 }
1253 
1254 
1255 /*
1256  * Read a HeapTuple as WAL logged by heap_insert, heap_update and heap_delete
1257  * (but not by heap_multi_insert) into a tuplebuf.
1258  *
1259  * The size 'len' and the pointer 'data' in the record need to be
1260  * computed outside as they are record specific.
1261  */
1262 static void
1264 {
1265  xl_heap_header xlhdr;
1266  int datalen = len - SizeOfHeapHeader;
1268 
1269  Assert(datalen >= 0);
1270 
1271  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1272  header = tuple->tuple.t_data;
1273 
1274  /* not a disk based tuple */
1276 
1277  /* we can only figure this out after reassembling the transactions */
1278  tuple->tuple.t_tableOid = InvalidOid;
1279 
1280  /* data is not stored aligned, copy to aligned storage */
1281  memcpy((char *) &xlhdr,
1282  data,
1283  SizeOfHeapHeader);
1284 
1285  memset(header, 0, SizeofHeapTupleHeader);
1286 
1287  memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
1288  data + SizeOfHeapHeader,
1289  datalen);
1290 
1291  header->t_infomask = xlhdr.t_infomask;
1292  header->t_infomask2 = xlhdr.t_infomask2;
1293  header->t_hoff = xlhdr.t_hoff;
1294 }
1295 
1296 /*
1297  * Check whether we are interested in this specific transaction.
1298  *
1299  * There can be several reasons we might not be interested in this
1300  * transaction:
1301  * 1) We might not be interested in decoding transactions up to this
1302  * LSN. This can happen because we previously decoded it and now just
1303  * are restarting or if we haven't assembled a consistent snapshot yet.
1304  * 2) The transaction happened in another database.
1305  * 3) The output plugin is not interested in the origin.
1306  * 4) We are doing fast-forwarding
1307  */
1308 static bool
1310  Oid txn_dbid, RepOriginId origin_id)
1311 {
1312  return (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
1313  (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) ||
1314  ctx->fast_forward || FilterByOrigin(ctx, origin_id));
1315 }
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1121
#define XLH_TRUNCATE_CASCADE
Definition: heapam_xlog.h:120
#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:99
#define XLOG_XACT_COMMIT
Definition: xact.h:148
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:172
#define XLOG_FPW_CHANGE
Definition: pg_control.h:75
struct ReorderBuffer * reorder
Definition: logical.h:45
bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: logical.c:1096
#define XLOG_XACT_PREPARE
Definition: xact.h:149
static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:901
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:59
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
uint32 TransactionId
Definition: c.h:587
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:385
#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:346
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:912
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
XLogRecPtr SnapBuildInitialConsistentPoint(SnapBuild *builder)
Definition: snapbuild.c:376
#define XLOG_XACT_INVALIDATIONS
Definition: xact.h:154
int64 TimestampTz
Definition: timestamp.h:39
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:53
TransactionId twophase_xid
Definition: xact.h:340
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1263
#define XLOG_INVALIDATIONS
Definition: standbydefs.h:36
OutputPluginCallbacks callbacks
Definition: logical.h:55
unsigned char uint8
Definition: c.h:439
#define XLOG_STANDBY_LOCK
Definition: standbydefs.h:34
TransactionId * subxacts
Definition: xact.h:332
#define XLH_INSERT_IS_SPECULATIVE
Definition: heapam_xlog.h:68
uint16 RepOriginId
Definition: xlogdefs.h:65
#define XLOG_RESTORE_POINT
Definition: pg_control.h:74
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
#define XLH_INSERT_ON_TOAST_RELATION
Definition: heapam_xlog.h:70
XLogRecPtr origin_lsn
Definition: xact.h:345
#define XLH_DELETE_CONTAINS_OLD
Definition: heapam_xlog.h:103
#define XLOG_HEAP_HOT_UPDATE
Definition: heapam_xlog.h:36
#define XLOG_HEAP2_PRUNE
Definition: heapam_xlog.h:54
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, XLogRecPtr initial_consistent_point, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn, char *gid, bool is_commit)
ReplicationSlotPersistentData data
Definition: slot.h:156
unsigned int Oid
Definition: postgres_ext.h:31
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:611
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:84
#define XLOG_END_OF_RECOVERY
Definition: pg_control.h:76
static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:960
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:380
#define XLOG_HEAP2_MULTI_INSERT
Definition: heapam_xlog.h:58
#define XLH_UPDATE_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:84
#define SizeOfHeapUpdate
Definition: heapam_xlog.h:227
char message[FLEXIBLE_ARRAY_MEMBER]
Definition: message.h:27
XLogRecPtr EndRecPtr
Definition: xlogreader.h:179
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:697
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:237
Oid relids[FLEXIBLE_ARRAY_MEMBER]
Definition: heapam_xlog.h:133
#define XLOG_NEXTOID
Definition: pg_control.h:70
static void DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:623
#define XLogRecGetData(decoder)
Definition: xlogreader.h:382
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1227
#define ERROR
Definition: elog.h:46
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:106
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:383
#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
SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER]
Definition: xact.h:255
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:678
void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
uint32 t_len
Definition: htup.h:64
#define SizeOfHeapDelete
Definition: heapam_xlog.h:115
XLogRecPtr endptr
Definition: decode.c:49
static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1027
void ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid)
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_XACT_ABORT_PREPARED
Definition: xact.h:152
#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:747
XLogRecPtr ReadRecPtr
Definition: xlogreader.h:178
HeapTupleData tuple
Definition: reorderbuffer.h:29
Oid t_tableOid
Definition: htup.h:66
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1079
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:60
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:1127
XLogReaderState * record
Definition: decode.c:50
#define XLOG_HEAP2_VACUUM
Definition: heapam_xlog.h:55
static void DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:195
#define XLOG_BACKUP_END
Definition: pg_control.h:72
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:377
union ReorderBufferChange::@97 data
RepOriginId origin_id
Definition: reorderbuffer.h:89
bool transactional
Definition: message.h:23
TransactionId * subxacts
Definition: xact.h:360
static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid dbId, RepOriginId origin_id)
Definition: decode.c:1309
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:2132
static void DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:441
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:379
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:153
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:2168
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
void UpdateDecodingStats(LogicalDecodingContext *ctx)
Definition: logical.c:1773
#define XLOG_NOOP
Definition: pg_control.h:69
LogicalDecodeFilterPrepareCB filter_prepare_cb
XLogRecPtr origin_lsn
Definition: xact.h:368
#define XLH_TRUNCATE_RESTART_SEQS
Definition: heapam_xlog.h:121
#define XLOG_HEAP_TRUNCATE
Definition: heapam_xlog.h:35
#define InvalidOid
Definition: postgres_ext.h:36
void ParsePrepareRecord(uint8 info, xl_xact_prepare *xlrec, xl_xact_parsed_prepare *parsed)
Definition: xactdesc.c:216
uint8 RmgrId
Definition: rmgr.h:11
TransactionId twophase_xid
Definition: xact.h:365
struct SnapBuild * snapshot_builder
Definition: logical.h:46
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:804
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
#define XLOG_HEAP_UPDATE
Definition: heapam_xlog.h:34
#define XLOG_XACT_ABORT
Definition: xact.h:150
#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:540
static bool FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: decode.c:588
#define XLOG_XACT_OPMASK
Definition: xact.h:158
TimestampTz xact_time
Definition: xact.h:353
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:1083
#define XLOG_HEAP_INPLACE
Definition: heapam_xlog.h:39
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:210
bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid, XLogRecPtr prepare_lsn, XLogRecPtr end_lsn, TimestampTz prepare_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
struct ReorderBufferChange::@97::@98 tp
#define XLOG_HEAP2_FREEZE_PAGE
Definition: heapam_xlog.h:56
XLogRecPtr origptr
Definition: decode.c:48
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:172
struct ReorderBufferChange::@97::@99 truncate
#define XLOG_FPI
Definition: pg_control.h:78
XLogReaderState * reader
Definition: logical.h:43
ReplicationSlot * slot
Definition: logical.h:40
#define XLOG_HEAP_LOCK
Definition: heapam_xlog.h:38
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define elog(elevel,...)
Definition: elog.h:232
Oid * ReorderBufferGetRelids(ReorderBuffer *rb, int nrelids)
int i
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1456
void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, char *gid)
#define XLOG_LOGICAL_MESSAGE
Definition: message.h:36
TransactionId oldestRunningXid
Definition: standbydefs.h:53
#define XLOG_RUNNING_XACTS
Definition: standbydefs.h:35
static void DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:501
void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define XLogRecGetTopXid(decoder)
Definition: xlogreader.h:381
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define XLH_UPDATE_CONTAINS_OLD
Definition: heapam_xlog.h:89
#define SHORTALIGN(LEN)
Definition: c.h:753
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:151
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
TimestampTz origin_timestamp
Definition: xact.h:369
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_prepare *parsed)
Definition: decode.c:772
LogicalDecodeFilterByOriginCB filter_by_origin_cb
#define XLOG_HEAP_DELETE
Definition: heapam_xlog.h:33
int nmsgs
Definition: xact.h:254
struct XLogRecordBuffer XLogRecordBuffer
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase)
Definition: decode.c:846
char twophase_gid[GIDSIZE]
Definition: xact.h:341
RmgrIds
Definition: rmgr.h:25
static void DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:395
#define SizeOfHeapHeader
Definition: heapam_xlog.h:151
char twophase_gid[GIDSIZE]
Definition: xact.h:366
#define XLogRecGetRmid(decoder)
Definition: xlogreader.h:378
#define XLOG_HEAP_CONFIRM
Definition: heapam_xlog.h:37
TimestampTz xact_time
Definition: xact.h:325
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:193
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:653
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)