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