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