Line data Source code
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"
32 : #include "access/xlog_internal.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"
39 : #include "replication/reorderbuffer.h"
40 : #include "replication/snapbuild.h"
41 : #include "storage/standbydefs.h"
42 :
43 : /* individual record(group)'s handlers */
44 : static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
45 : static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
46 : static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
47 : static void DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
48 : static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
49 : static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
50 :
51 : static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
52 : xl_xact_parsed_commit *parsed, TransactionId xid,
53 : bool two_phase);
54 : static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
55 : xl_xact_parsed_abort *parsed, TransactionId xid,
56 : bool two_phase);
57 : static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
58 : xl_xact_parsed_prepare *parsed);
59 :
60 :
61 : /* common function to decode tuples */
62 : static void DecodeXLogTuple(char *data, Size len, HeapTuple tuple);
63 :
64 : /* helper functions for decoding transactions */
65 : static inline bool FilterPrepare(LogicalDecodingContext *ctx,
66 : TransactionId xid, const char *gid);
67 : static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx,
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 : */
87 : void
88 4959038 : LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
89 : {
90 : XLogRecordBuffer buf;
91 : TransactionId txid;
92 : RmgrData rmgr;
93 :
94 4959038 : buf.origptr = ctx->reader->ReadRecPtr;
95 4959038 : buf.endptr = ctx->reader->EndRecPtr;
96 4959038 : buf.record = record;
97 :
98 4959038 : 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 4959038 : if (TransactionIdIsValid(txid))
106 : {
107 1368 : ReorderBufferAssignChild(ctx->reorder,
108 : txid,
109 1368 : XLogRecGetXid(record),
110 : buf.origptr);
111 : }
112 :
113 4959038 : rmgr = GetRmgr(XLogRecGetRmid(record));
114 :
115 4959038 : if (rmgr.rm_decode != NULL)
116 3856072 : rmgr.rm_decode(ctx, &buf);
117 : else
118 : {
119 : /* just deal with xid, and done */
120 1102966 : ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(record),
121 : buf.origptr);
122 : }
123 4959028 : }
124 :
125 : /*
126 : * Handle rmgr XLOG_ID records for LogicalDecodingProcessRecord().
127 : */
128 : void
129 11598 : xlog_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
130 : {
131 11598 : SnapBuild *builder = ctx->snapshot_builder;
132 11598 : uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
133 :
134 11598 : ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(buf->record),
135 : buf->origptr);
136 :
137 11598 : switch (info)
138 : {
139 : /* this is also used in END_OF_RECOVERY checkpoints */
140 76 : case XLOG_CHECKPOINT_SHUTDOWN:
141 : case XLOG_END_OF_RECOVERY:
142 76 : SnapBuildSerializationPoint(builder, buf->origptr);
143 :
144 76 : break;
145 126 : case XLOG_CHECKPOINT_ONLINE:
146 :
147 : /*
148 : * a RUNNING_XACTS record will have been logged near to this, we
149 : * can restart from there.
150 : */
151 126 : break;
152 2 : case XLOG_PARAMETER_CHANGE:
153 : {
154 2 : xl_parameter_change *xlrec =
155 2 : (xl_parameter_change *) XLogRecGetData(buf->record);
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 2 : 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 : */
174 : Assert(RecoveryInProgress());
175 0 : ereport(ERROR,
176 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
177 : errmsg("logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary")));
178 : }
179 2 : break;
180 : }
181 11394 : case XLOG_NOOP:
182 : case XLOG_NEXTOID:
183 : case XLOG_SWITCH:
184 : case XLOG_BACKUP_END:
185 : case XLOG_RESTORE_POINT:
186 : case XLOG_FPW_CHANGE:
187 : case XLOG_FPI_FOR_HINT:
188 : case XLOG_FPI:
189 : case XLOG_OVERWRITE_CONTRECORD:
190 : case XLOG_CHECKPOINT_REDO:
191 11394 : break;
192 0 : default:
193 0 : elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
194 : }
195 11598 : }
196 :
197 : /*
198 : * Handle rmgr XACT_ID records for LogicalDecodingProcessRecord().
199 : */
200 : void
201 15748 : xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
202 : {
203 15748 : SnapBuild *builder = ctx->snapshot_builder;
204 15748 : ReorderBuffer *reorder = ctx->reorder;
205 15748 : XLogReaderState *r = buf->record;
206 15748 : uint8 info = XLogRecGetInfo(r) & XLOG_XACT_OPMASK;
207 :
208 : /*
209 : * If the snapshot isn't yet fully built, we cannot decode anything, so
210 : * bail out.
211 : */
212 15748 : if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
213 28 : return;
214 :
215 15720 : switch (info)
216 : {
217 5410 : case XLOG_XACT_COMMIT:
218 : case XLOG_XACT_COMMIT_PREPARED:
219 : {
220 : xl_xact_commit *xlrec;
221 : xl_xact_parsed_commit parsed;
222 : TransactionId xid;
223 5410 : bool two_phase = false;
224 :
225 5410 : xlrec = (xl_xact_commit *) XLogRecGetData(r);
226 5410 : ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
227 :
228 5410 : if (!TransactionIdIsValid(parsed.twophase_xid))
229 5212 : xid = XLogRecGetXid(r);
230 : else
231 198 : 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 5410 : if (info == XLOG_XACT_COMMIT_PREPARED)
239 198 : two_phase = !(FilterPrepare(ctx, xid,
240 198 : parsed.twophase_gid));
241 :
242 5410 : DecodeCommit(ctx, buf, &parsed, xid, two_phase);
243 5404 : break;
244 : }
245 228 : case XLOG_XACT_ABORT:
246 : case XLOG_XACT_ABORT_PREPARED:
247 : {
248 : xl_xact_abort *xlrec;
249 : xl_xact_parsed_abort parsed;
250 : TransactionId xid;
251 228 : bool two_phase = false;
252 :
253 228 : xlrec = (xl_xact_abort *) XLogRecGetData(r);
254 228 : ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
255 :
256 228 : if (!TransactionIdIsValid(parsed.twophase_xid))
257 152 : xid = XLogRecGetXid(r);
258 : else
259 76 : 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 228 : if (info == XLOG_XACT_ABORT_PREPARED)
267 76 : two_phase = !(FilterPrepare(ctx, xid,
268 76 : parsed.twophase_gid));
269 :
270 228 : DecodeAbort(ctx, buf, &parsed, xid, two_phase);
271 228 : break;
272 : }
273 266 : case XLOG_XACT_ASSIGNMENT:
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 266 : break;
281 9504 : case XLOG_XACT_INVALIDATIONS:
282 : {
283 : TransactionId xid;
284 : xl_xact_invals *invals;
285 :
286 9504 : xid = XLogRecGetXid(r);
287 9504 : 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 9504 : if (TransactionIdIsValid(xid))
295 : {
296 9500 : if (!ctx->fast_forward)
297 9468 : ReorderBufferAddInvalidations(reorder, xid,
298 : buf->origptr,
299 9468 : invals->nmsgs,
300 9468 : invals->msgs);
301 9500 : ReorderBufferXidSetCatalogChanges(ctx->reorder, xid,
302 : buf->origptr);
303 : }
304 4 : else if (!ctx->fast_forward)
305 4 : ReorderBufferImmediateInvalidation(ctx->reorder,
306 4 : invals->nmsgs,
307 4 : invals->msgs);
308 :
309 9504 : break;
310 : }
311 312 : case XLOG_XACT_PREPARE:
312 : {
313 : xl_xact_parsed_prepare parsed;
314 : xl_xact_prepare *xlrec;
315 :
316 : /* ok, parse it */
317 312 : xlrec = (xl_xact_prepare *) XLogRecGetData(r);
318 312 : ParsePrepareRecord(XLogRecGetInfo(buf->record),
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 312 : if (FilterPrepare(ctx, parsed.twophase_xid,
327 : parsed.twophase_gid))
328 : {
329 28 : ReorderBufferProcessXid(reorder, parsed.twophase_xid,
330 : buf->origptr);
331 28 : 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 284 : DecodePrepare(ctx, buf, &parsed);
348 284 : break;
349 : }
350 0 : default:
351 0 : elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
352 : }
353 : }
354 :
355 : /*
356 : * Handle rmgr STANDBY_ID records for LogicalDecodingProcessRecord().
357 : */
358 : void
359 7028 : standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
360 : {
361 7028 : SnapBuild *builder = ctx->snapshot_builder;
362 7028 : XLogReaderState *r = buf->record;
363 7028 : uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
364 :
365 7028 : ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
366 :
367 7028 : switch (info)
368 : {
369 2548 : case XLOG_RUNNING_XACTS:
370 : {
371 2548 : xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r);
372 :
373 2548 : 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 : */
384 2544 : ReorderBufferAbortOld(ctx->reorder, running->oldestRunningXid);
385 : }
386 2544 : break;
387 4476 : case XLOG_STANDBY_LOCK:
388 4476 : break;
389 4 : case XLOG_INVALIDATIONS:
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 4 : break;
396 0 : default:
397 0 : elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
398 : }
399 7024 : }
400 :
401 : /*
402 : * Handle rmgr HEAP2_ID records for LogicalDecodingProcessRecord().
403 : */
404 : void
405 62180 : heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
406 : {
407 62180 : uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
408 62180 : TransactionId xid = XLogRecGetXid(buf->record);
409 62180 : SnapBuild *builder = ctx->snapshot_builder;
410 :
411 62180 : 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 : */
417 62180 : if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
418 62162 : ctx->fast_forward)
419 260 : return;
420 :
421 61920 : switch (info)
422 : {
423 11336 : case XLOG_HEAP2_MULTI_INSERT:
424 11336 : if (SnapBuildProcessChange(builder, xid, buf->origptr))
425 11336 : DecodeMultiInsert(ctx, buf);
426 11336 : break;
427 45700 : case XLOG_HEAP2_NEW_CID:
428 : {
429 : xl_heap_new_cid *xlrec;
430 :
431 45700 : xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
432 45700 : SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
433 :
434 45700 : break;
435 : }
436 180 : case XLOG_HEAP2_REWRITE:
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 180 : break;
444 :
445 : /*
446 : * Everything else here is just low level physical stuff we're not
447 : * interested in.
448 : */
449 4704 : case XLOG_HEAP2_PRUNE_ON_ACCESS:
450 : case XLOG_HEAP2_PRUNE_VACUUM_SCAN:
451 : case XLOG_HEAP2_PRUNE_VACUUM_CLEANUP:
452 : case XLOG_HEAP2_VISIBLE:
453 : case XLOG_HEAP2_LOCK_UPDATED:
454 4704 : break;
455 0 : default:
456 0 : elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
457 : }
458 : }
459 :
460 : /*
461 : * Handle rmgr HEAP_ID records for LogicalDecodingProcessRecord().
462 : */
463 : void
464 3759404 : heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
465 : {
466 3759404 : uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
467 3759404 : TransactionId xid = XLogRecGetXid(buf->record);
468 3759404 : SnapBuild *builder = ctx->snapshot_builder;
469 :
470 3759404 : 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 : */
476 3759404 : if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
477 3759396 : ctx->fast_forward)
478 188 : return;
479 :
480 3759216 : switch (info)
481 : {
482 2392334 : case XLOG_HEAP_INSERT:
483 2392334 : if (SnapBuildProcessChange(builder, xid, buf->origptr))
484 2392334 : DecodeInsert(ctx, buf);
485 2392334 : 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 : */
492 414292 : case XLOG_HEAP_HOT_UPDATE:
493 : case XLOG_HEAP_UPDATE:
494 414292 : if (SnapBuildProcessChange(builder, xid, buf->origptr))
495 414292 : DecodeUpdate(ctx, buf);
496 414292 : break;
497 :
498 533874 : case XLOG_HEAP_DELETE:
499 533874 : if (SnapBuildProcessChange(builder, xid, buf->origptr))
500 533874 : DecodeDelete(ctx, buf);
501 533874 : break;
502 :
503 106 : case XLOG_HEAP_TRUNCATE:
504 106 : if (SnapBuildProcessChange(builder, xid, buf->origptr))
505 106 : DecodeTruncate(ctx, buf);
506 106 : break;
507 :
508 1980 : case XLOG_HEAP_INPLACE:
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 1980 : break;
526 :
527 35832 : case XLOG_HEAP_CONFIRM:
528 35832 : if (SnapBuildProcessChange(builder, xid, buf->origptr))
529 35832 : DecodeSpecConfirm(ctx, buf);
530 35832 : break;
531 :
532 380798 : case XLOG_HEAP_LOCK:
533 : /* we don't care about row level locks for now */
534 380798 : break;
535 :
536 0 : default:
537 0 : 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 : */
546 : static inline bool
547 586 : FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid,
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 586 : if (!ctx->twophase)
557 26 : return true;
558 :
559 : /*
560 : * The filter_prepare callback is optional. When not supplied, all
561 : * prepared transactions should go through.
562 : */
563 560 : if (ctx->callbacks.filter_prepare_cb == NULL)
564 272 : return false;
565 :
566 288 : return filter_prepare_cb_wrapper(ctx, xid, gid);
567 : }
568 :
569 : static inline bool
570 3371964 : FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
571 : {
572 3371964 : if (ctx->callbacks.filter_by_origin_cb == NULL)
573 42 : return false;
574 :
575 3371922 : return filter_by_origin_cb_wrapper(ctx, origin_id);
576 : }
577 :
578 : /*
579 : * Handle rmgr LOGICALMSG_ID records for LogicalDecodingProcessRecord().
580 : */
581 : void
582 114 : logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
583 : {
584 114 : SnapBuild *builder = ctx->snapshot_builder;
585 114 : XLogReaderState *r = buf->record;
586 114 : TransactionId xid = XLogRecGetXid(r);
587 114 : uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
588 114 : RepOriginId origin_id = XLogRecGetOrigin(r);
589 114 : Snapshot snapshot = NULL;
590 : xl_logical_message *message;
591 :
592 114 : if (info != XLOG_LOGICAL_MESSAGE)
593 0 : elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
594 :
595 114 : ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
596 :
597 : /* If we don't have snapshot, there is no point in decoding messages */
598 114 : if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
599 0 : return;
600 :
601 114 : message = (xl_logical_message *) XLogRecGetData(r);
602 :
603 224 : if (message->dbId != ctx->slot->data.database ||
604 110 : FilterByOrigin(ctx, origin_id))
605 8 : return;
606 :
607 106 : if (message->transactional &&
608 78 : !SnapBuildProcessChange(builder, xid, buf->origptr))
609 0 : return;
610 134 : else if (!message->transactional &&
611 56 : (SnapBuildCurrentState(builder) != SNAPBUILD_CONSISTENT ||
612 28 : SnapBuildXactNeedsSkip(builder, buf->origptr)))
613 8 : 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 98 : 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 4 : if (!message->transactional)
630 4 : ctx->processing_required = true;
631 :
632 4 : 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 94 : if (!message->transactional)
644 16 : snapshot = SnapBuildGetOrBuildSnapshot(builder);
645 :
646 94 : ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
647 94 : message->transactional,
648 94 : message->message, /* first part of message is
649 : * prefix */
650 : message->message_size,
651 94 : 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 : */
662 : static void
663 5410 : DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
664 : xl_xact_parsed_commit *parsed, TransactionId xid,
665 : bool two_phase)
666 : {
667 5410 : XLogRecPtr origin_lsn = InvalidXLogRecPtr;
668 5410 : TimestampTz commit_time = parsed->xact_time;
669 5410 : RepOriginId origin_id = XLogRecGetOrigin(buf->record);
670 : int i;
671 :
672 5410 : if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
673 : {
674 134 : origin_lsn = parsed->origin_lsn;
675 134 : commit_time = parsed->origin_timestamp;
676 : }
677 :
678 5410 : 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 5410 : if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
700 : {
701 4984 : for (i = 0; i < parsed->nsubxacts; i++)
702 : {
703 1968 : ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
704 : }
705 3016 : ReorderBufferForget(ctx->reorder, xid, buf->origptr);
706 :
707 3016 : return;
708 : }
709 :
710 : /* tell the reorderbuffer about the surviving subtransactions */
711 2928 : for (i = 0; i < parsed->nsubxacts; i++)
712 : {
713 534 : 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 2394 : if (two_phase)
722 : {
723 62 : ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
724 62 : SnapBuildGetTwoPhaseAt(ctx->snapshot_builder),
725 : commit_time, origin_id, origin_lsn,
726 62 : parsed->twophase_gid, true);
727 : }
728 : else
729 : {
730 2332 : 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 : */
740 2388 : UpdateDecodingStats(ctx);
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 : */
758 : static void
759 284 : DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
760 : xl_xact_parsed_prepare *parsed)
761 : {
762 284 : SnapBuild *builder = ctx->snapshot_builder;
763 284 : XLogRecPtr origin_lsn = parsed->origin_lsn;
764 284 : TimestampTz prepare_time = parsed->xact_time;
765 284 : RepOriginId origin_id = XLogRecGetOrigin(buf->record);
766 : int i;
767 284 : TransactionId xid = parsed->twophase_xid;
768 :
769 284 : if (parsed->origin_timestamp != 0)
770 16 : 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 284 : if (!ReorderBufferRememberPrepareInfo(ctx->reorder, xid, buf->origptr,
777 : buf->endptr, prepare_time, origin_id,
778 : origin_lsn))
779 0 : return;
780 :
781 : /* We can't start streaming unless a consistent state is reached. */
782 284 : if (SnapBuildCurrentState(builder) < SNAPBUILD_CONSISTENT)
783 : {
784 6 : ReorderBufferSkipPrepare(ctx->reorder, xid);
785 6 : 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 278 : if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
801 : {
802 198 : ReorderBufferSkipPrepare(ctx->reorder, xid);
803 198 : ReorderBufferInvalidate(ctx->reorder, xid, buf->origptr);
804 198 : return;
805 : }
806 :
807 : /* Tell the reorderbuffer about the surviving subtransactions. */
808 82 : for (i = 0; i < parsed->nsubxacts; i++)
809 : {
810 2 : ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
811 : buf->origptr, buf->endptr);
812 : }
813 :
814 : /* replay actions of all transaction + subtransactions in order */
815 80 : 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 : */
823 80 : UpdateDecodingStats(ctx);
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 : */
833 : static void
834 228 : DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
835 : xl_xact_parsed_abort *parsed, TransactionId xid,
836 : bool two_phase)
837 : {
838 : int i;
839 228 : XLogRecPtr origin_lsn = InvalidXLogRecPtr;
840 228 : TimestampTz abort_time = parsed->xact_time;
841 228 : RepOriginId origin_id = XLogRecGetOrigin(buf->record);
842 : bool skip_xact;
843 :
844 228 : if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
845 : {
846 8 : origin_lsn = parsed->origin_lsn;
847 8 : 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 228 : 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 228 : if (two_phase && !skip_xact)
862 : {
863 20 : ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
864 : InvalidXLogRecPtr,
865 : abort_time, origin_id, origin_lsn,
866 20 : parsed->twophase_gid, false);
867 : }
868 : else
869 : {
870 220 : for (i = 0; i < parsed->nsubxacts; i++)
871 : {
872 12 : ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
873 12 : buf->record->EndRecPtr, abort_time);
874 : }
875 :
876 208 : ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr,
877 : abort_time);
878 : }
879 :
880 : /* update the decoding stats */
881 228 : UpdateDecodingStats(ctx);
882 228 : }
883 :
884 : /*
885 : * Parse XLOG_HEAP_INSERT (not MULTI_INSERT!) records into tuplebufs.
886 : *
887 : * Inserts can contain the new tuple.
888 : */
889 : static void
890 2392334 : DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
891 : {
892 : Size datalen;
893 : char *tupledata;
894 : Size tuplelen;
895 2392334 : XLogReaderState *r = buf->record;
896 : xl_heap_insert *xlrec;
897 : ReorderBufferChange *change;
898 : RelFileLocator target_locator;
899 :
900 2392334 : 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 2392334 : if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
907 7258 : return;
908 :
909 : /* only interested in our database */
910 2385110 : XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
911 2385110 : if (target_locator.dbOid != ctx->slot->data.database)
912 0 : return;
913 :
914 : /* output plugin doesn't look for this origin, no need to queue */
915 2385110 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
916 34 : return;
917 :
918 2385076 : change = ReorderBufferGetChange(ctx->reorder);
919 2385076 : if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
920 2349244 : change->action = REORDER_BUFFER_CHANGE_INSERT;
921 : else
922 35832 : change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT;
923 2385076 : change->origin_id = XLogRecGetOrigin(r);
924 :
925 2385076 : memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
926 :
927 2385076 : tupledata = XLogRecGetBlockData(r, 0, &datalen);
928 2385076 : tuplelen = datalen - SizeOfHeapHeader;
929 :
930 2385076 : change->data.tp.newtuple =
931 2385076 : ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
932 :
933 2385076 : DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
934 :
935 2385076 : change->data.tp.clear_toast_afterwards = true;
936 :
937 2385076 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
938 : change,
939 2385076 : xlrec->flags & XLH_INSERT_ON_TOAST_RELATION);
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 : */
948 : static void
949 414292 : DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
950 : {
951 414292 : XLogReaderState *r = buf->record;
952 : xl_heap_update *xlrec;
953 : ReorderBufferChange *change;
954 : char *data;
955 : RelFileLocator target_locator;
956 :
957 414292 : xlrec = (xl_heap_update *) XLogRecGetData(r);
958 :
959 : /* only interested in our database */
960 414292 : XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
961 414292 : if (target_locator.dbOid != ctx->slot->data.database)
962 66 : return;
963 :
964 : /* output plugin doesn't look for this origin, no need to queue */
965 414278 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
966 52 : return;
967 :
968 414226 : change = ReorderBufferGetChange(ctx->reorder);
969 414226 : change->action = REORDER_BUFFER_CHANGE_UPDATE;
970 414226 : change->origin_id = XLogRecGetOrigin(r);
971 414226 : memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
972 :
973 414226 : if (xlrec->flags & XLH_UPDATE_CONTAINS_NEW_TUPLE)
974 : {
975 : Size datalen;
976 : Size tuplelen;
977 :
978 410860 : data = XLogRecGetBlockData(r, 0, &datalen);
979 :
980 410860 : tuplelen = datalen - SizeOfHeapHeader;
981 :
982 410860 : change->data.tp.newtuple =
983 410860 : ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
984 :
985 410860 : DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
986 : }
987 :
988 414226 : if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
989 : {
990 : Size datalen;
991 : Size tuplelen;
992 :
993 : /* caution, remaining data in record is not aligned */
994 574 : data = XLogRecGetData(r) + SizeOfHeapUpdate;
995 574 : datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
996 574 : tuplelen = datalen - SizeOfHeapHeader;
997 :
998 574 : change->data.tp.oldtuple =
999 574 : ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1000 :
1001 574 : DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
1002 : }
1003 :
1004 414226 : change->data.tp.clear_toast_afterwards = true;
1005 :
1006 414226 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
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 : */
1015 : static void
1016 533874 : DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
1017 : {
1018 533874 : XLogReaderState *r = buf->record;
1019 : xl_heap_delete *xlrec;
1020 : ReorderBufferChange *change;
1021 : RelFileLocator target_locator;
1022 :
1023 533874 : xlrec = (xl_heap_delete *) XLogRecGetData(r);
1024 :
1025 : /* only interested in our database */
1026 533874 : XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
1027 533874 : if (target_locator.dbOid != ctx->slot->data.database)
1028 62 : return;
1029 :
1030 : /* output plugin doesn't look for this origin, no need to queue */
1031 533824 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1032 12 : return;
1033 :
1034 533812 : change = ReorderBufferGetChange(ctx->reorder);
1035 :
1036 533812 : if (xlrec->flags & XLH_DELETE_IS_SUPER)
1037 0 : change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT;
1038 : else
1039 533812 : change->action = REORDER_BUFFER_CHANGE_DELETE;
1040 :
1041 533812 : change->origin_id = XLogRecGetOrigin(r);
1042 :
1043 533812 : memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
1044 :
1045 : /* old primary key stored */
1046 533812 : if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
1047 : {
1048 411444 : Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
1049 411444 : Size tuplelen = datalen - SizeOfHeapHeader;
1050 :
1051 : Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
1052 :
1053 411444 : change->data.tp.oldtuple =
1054 411444 : ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
1055 :
1056 411444 : DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
1057 : datalen, change->data.tp.oldtuple);
1058 : }
1059 :
1060 533812 : change->data.tp.clear_toast_afterwards = true;
1061 :
1062 533812 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
1063 : change, false);
1064 : }
1065 :
1066 : /*
1067 : * Parse XLOG_HEAP_TRUNCATE from wal
1068 : */
1069 : static void
1070 106 : DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
1071 : {
1072 106 : XLogReaderState *r = buf->record;
1073 : xl_heap_truncate *xlrec;
1074 : ReorderBufferChange *change;
1075 :
1076 106 : xlrec = (xl_heap_truncate *) XLogRecGetData(r);
1077 :
1078 : /* only interested in our database */
1079 106 : if (xlrec->dbId != ctx->slot->data.database)
1080 0 : return;
1081 :
1082 : /* output plugin doesn't look for this origin, no need to queue */
1083 106 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1084 0 : return;
1085 :
1086 106 : change = ReorderBufferGetChange(ctx->reorder);
1087 106 : change->action = REORDER_BUFFER_CHANGE_TRUNCATE;
1088 106 : change->origin_id = XLogRecGetOrigin(r);
1089 106 : if (xlrec->flags & XLH_TRUNCATE_CASCADE)
1090 2 : change->data.truncate.cascade = true;
1091 106 : if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
1092 4 : change->data.truncate.restart_seqs = true;
1093 106 : change->data.truncate.nrelids = xlrec->nrelids;
1094 212 : change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
1095 106 : xlrec->nrelids);
1096 106 : memcpy(change->data.truncate.relids, xlrec->relids,
1097 106 : xlrec->nrelids * sizeof(Oid));
1098 106 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
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 : */
1107 : static void
1108 11336 : DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
1109 : {
1110 11336 : 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 11336 : 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 11336 : if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
1125 11302 : return;
1126 :
1127 : /* only interested in our database */
1128 136 : XLogRecGetBlockTag(r, 0, &rlocator, NULL, NULL);
1129 136 : if (rlocator.dbOid != ctx->slot->data.database)
1130 102 : return;
1131 :
1132 : /* output plugin doesn't look for this origin, no need to queue */
1133 34 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1134 0 : 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 34 : tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
1141 : Assert(tupledata != NULL);
1142 :
1143 34 : data = tupledata;
1144 2140 : 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 2106 : change = ReorderBufferGetChange(ctx->reorder);
1153 2106 : change->action = REORDER_BUFFER_CHANGE_INSERT;
1154 2106 : change->origin_id = XLogRecGetOrigin(r);
1155 :
1156 2106 : memcpy(&change->data.tp.rlocator, &rlocator, sizeof(RelFileLocator));
1157 :
1158 2106 : xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
1159 2106 : data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
1160 2106 : datalen = xlhdr->datalen;
1161 :
1162 2106 : change->data.tp.newtuple =
1163 2106 : ReorderBufferGetTupleBuf(ctx->reorder, datalen);
1164 :
1165 2106 : tuple = change->data.tp.newtuple;
1166 2106 : header = tuple->t_data;
1167 :
1168 : /* not a disk based tuple */
1169 2106 : ItemPointerSetInvalid(&tuple->t_self);
1170 :
1171 : /*
1172 : * We can only figure this out after reassembling the transactions.
1173 : */
1174 2106 : tuple->t_tableOid = InvalidOid;
1175 :
1176 2106 : tuple->t_len = datalen + SizeofHeapTupleHeader;
1177 :
1178 2106 : memset(header, 0, SizeofHeapTupleHeader);
1179 :
1180 2106 : memcpy((char *) tuple->t_data + SizeofHeapTupleHeader,
1181 : (char *) data,
1182 : datalen);
1183 2106 : header->t_infomask = xlhdr->t_infomask;
1184 2106 : header->t_infomask2 = xlhdr->t_infomask2;
1185 2106 : 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 2106 : if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
1193 386 : (i + 1) == xlrec->ntuples)
1194 24 : change->data.tp.clear_toast_afterwards = true;
1195 : else
1196 2082 : change->data.tp.clear_toast_afterwards = false;
1197 :
1198 2106 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
1199 : buf->origptr, change, false);
1200 :
1201 : /* move to the next xl_multi_insert_tuple entry */
1202 2106 : 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 : */
1213 : static void
1214 35832 : DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
1215 : {
1216 35832 : XLogReaderState *r = buf->record;
1217 : ReorderBufferChange *change;
1218 : RelFileLocator target_locator;
1219 :
1220 : /* only interested in our database */
1221 35832 : XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
1222 35832 : if (target_locator.dbOid != ctx->slot->data.database)
1223 0 : return;
1224 :
1225 : /* output plugin doesn't look for this origin, no need to queue */
1226 35832 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1227 0 : return;
1228 :
1229 35832 : change = ReorderBufferGetChange(ctx->reorder);
1230 35832 : change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM;
1231 35832 : change->origin_id = XLogRecGetOrigin(r);
1232 :
1233 35832 : memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
1234 :
1235 35832 : change->data.tp.clear_toast_afterwards = true;
1236 :
1237 35832 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
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 : */
1249 : static void
1250 3207954 : DecodeXLogTuple(char *data, Size len, HeapTuple tuple)
1251 : {
1252 : xl_heap_header xlhdr;
1253 3207954 : int datalen = len - SizeOfHeapHeader;
1254 : HeapTupleHeader header;
1255 :
1256 : Assert(datalen >= 0);
1257 :
1258 3207954 : tuple->t_len = datalen + SizeofHeapTupleHeader;
1259 3207954 : header = tuple->t_data;
1260 :
1261 : /* not a disk based tuple */
1262 3207954 : ItemPointerSetInvalid(&tuple->t_self);
1263 :
1264 : /* we can only figure this out after reassembling the transactions */
1265 3207954 : tuple->t_tableOid = InvalidOid;
1266 :
1267 : /* data is not stored aligned, copy to aligned storage */
1268 3207954 : memcpy((char *) &xlhdr,
1269 : data,
1270 : SizeOfHeapHeader);
1271 :
1272 3207954 : memset(header, 0, SizeofHeapTupleHeader);
1273 :
1274 3207954 : memcpy(((char *) tuple->t_data) + SizeofHeapTupleHeader,
1275 3207954 : data + SizeOfHeapHeader,
1276 : datalen);
1277 :
1278 3207954 : header->t_infomask = xlhdr.t_infomask;
1279 3207954 : header->t_infomask2 = xlhdr.t_infomask2;
1280 3207954 : header->t_hoff = xlhdr.t_hoff;
1281 3207954 : }
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 : */
1295 : static bool
1296 5916 : DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
1297 : Oid txn_dbid, RepOriginId origin_id)
1298 : {
1299 5916 : if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
1300 5302 : (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) ||
1301 2670 : FilterByOrigin(ctx, origin_id))
1302 3302 : 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 2614 : if (ctx->fast_forward)
1310 : {
1311 38 : ctx->processing_required = true;
1312 38 : return true;
1313 : }
1314 :
1315 2576 : return false;
1316 : }
|
