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