Age Owner Branch data TLA 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-2025, 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
3943 heikki.linnakangas@i 88 :CBC 1910252 : LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
89 : : {
90 : : XLogRecordBuffer buf;
91 : : TransactionId txid;
92 : : RmgrData rmgr;
93 : :
4205 rhaas@postgresql.org 94 : 1910252 : buf.origptr = ctx->reader->ReadRecPtr;
95 : 1910252 : buf.endptr = ctx->reader->EndRecPtr;
3943 heikki.linnakangas@i 96 : 1910252 : buf.record = record;
97 : :
1874 akapila@postgresql.o 98 : 1910252 : 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 [ + + ]: 1910252 : if (TransactionIdIsValid(txid))
106 : : {
107 : 643 : ReorderBufferAssignChild(ctx->reorder,
108 : : txid,
1268 tmunro@postgresql.or 109 : 643 : XLogRecGetXid(record),
110 : : buf.origptr);
111 : : }
112 : :
1249 jdavis@postgresql.or 113 : 1910252 : rmgr = GetRmgr(XLogRecGetRmid(record));
114 : :
115 [ + + ]: 1910252 : if (rmgr.rm_decode != NULL)
116 : 1462787 : rmgr.rm_decode(ctx, &buf);
117 : : else
118 : : {
119 : : /* just deal with xid, and done */
1326 120 : 447465 : ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(record),
121 : : buf.origptr);
122 : : }
4205 rhaas@postgresql.org 123 : 1910243 : }
124 : :
125 : : /*
126 : : * Handle rmgr XLOG_ID records for LogicalDecodingProcessRecord().
127 : : */
128 : : void
1326 jdavis@postgresql.or 129 : 6096 : xlog_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
130 : : {
4205 rhaas@postgresql.org 131 : 6096 : SnapBuild *builder = ctx->snapshot_builder;
3943 heikki.linnakangas@i 132 : 6096 : uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
133 : :
3472 andres@anarazel.de 134 : 6096 : ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(buf->record),
135 : : buf->origptr);
136 : :
4205 rhaas@postgresql.org 137 [ + + + + : 6096 : switch (info)
- ]
138 : : {
139 : : /* this is also used in END_OF_RECOVERY checkpoints */
140 : 57 : case XLOG_CHECKPOINT_SHUTDOWN:
141 : : case XLOG_END_OF_RECOVERY:
142 : 57 : SnapBuildSerializationPoint(builder, buf->origptr);
143 : :
144 : 57 : break;
145 : 60 : 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 : 60 : break;
882 andres@anarazel.de 152 : 14 : case XLOG_PARAMETER_CHANGE:
153 : : {
154 : 14 : xl_parameter_change *xlrec =
841 tgl@sss.pgh.pa.us 155 : 14 : (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 : : */
882 andres@anarazel.de 167 [ - + ]: 14 : 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 : : */
882 andres@anarazel.de 174 [ # # ]:UBC 0 : 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 : : }
882 andres@anarazel.de 179 :CBC 14 : break;
180 : : }
4205 rhaas@postgresql.org 181 : 5965 : 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 : 5965 : break;
4205 rhaas@postgresql.org 192 :UBC 0 : default:
193 [ # # ]: 0 : elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
194 : : }
4205 rhaas@postgresql.org 195 :CBC 6096 : }
196 : :
197 : : /*
198 : : * Handle rmgr XACT_ID records for LogicalDecodingProcessRecord().
199 : : */
200 : : void
1326 jdavis@postgresql.or 201 : 8628 : xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
202 : : {
4141 bruce@momjian.us 203 : 8628 : SnapBuild *builder = ctx->snapshot_builder;
204 : 8628 : ReorderBuffer *reorder = ctx->reorder;
3943 heikki.linnakangas@i 205 : 8628 : XLogReaderState *r = buf->record;
3828 andres@anarazel.de 206 : 8628 : 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 : : */
1874 akapila@postgresql.o 212 [ + + ]: 8628 : if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
4205 rhaas@postgresql.org 213 : 15 : return;
214 : :
215 [ + + + + : 8613 : switch (info)
+ - ]
216 : : {
217 : 3037 : 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;
1706 akapila@postgresql.o 223 : 3037 : bool two_phase = false;
224 : :
3828 andres@anarazel.de 225 : 3037 : xlrec = (xl_xact_commit *) XLogRecGetData(r);
226 : 3037 : ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
227 : :
228 [ + + ]: 3037 : if (!TransactionIdIsValid(parsed.twophase_xid))
229 : 2937 : xid = XLogRecGetXid(r);
230 : : else
231 : 100 : 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 : : */
1706 akapila@postgresql.o 238 [ + + ]: 3037 : if (info == XLOG_XACT_COMMIT_PREPARED)
1621 239 : 100 : two_phase = !(FilterPrepare(ctx, xid,
240 : 100 : parsed.twophase_gid));
241 : :
1706 242 : 3037 : DecodeCommit(ctx, buf, &parsed, xid, two_phase);
4205 rhaas@postgresql.org 243 : 3030 : break;
244 : : }
245 : 153 : 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;
1706 akapila@postgresql.o 251 : 153 : bool two_phase = false;
252 : :
3828 andres@anarazel.de 253 : 153 : xlrec = (xl_xact_abort *) XLogRecGetData(r);
254 : 153 : ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
255 : :
256 [ + + ]: 153 : if (!TransactionIdIsValid(parsed.twophase_xid))
257 : 115 : xid = XLogRecGetXid(r);
258 : : else
259 : 38 : 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 : : */
1706 akapila@postgresql.o 266 [ + + ]: 153 : if (info == XLOG_XACT_ABORT_PREPARED)
1621 267 : 38 : two_phase = !(FilterPrepare(ctx, xid,
268 : 38 : parsed.twophase_gid));
269 : :
1706 270 : 153 : DecodeAbort(ctx, buf, &parsed, xid, two_phase);
4205 rhaas@postgresql.org 271 : 153 : break;
272 : : }
273 : 108 : 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 : : */
1874 akapila@postgresql.o 280 : 108 : break;
1871 281 : 5155 : case XLOG_XACT_INVALIDATIONS:
282 : : {
283 : : TransactionId xid;
284 : : xl_xact_invals *invals;
285 : :
286 : 5155 : xid = XLogRecGetXid(r);
287 : 5155 : 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 [ + + ]: 5155 : if (TransactionIdIsValid(xid))
295 : : {
296 [ + + ]: 5146 : if (!ctx->fast_forward)
297 : 5113 : ReorderBufferAddInvalidations(reorder, xid,
298 : : buf->origptr,
299 : 5113 : invals->nmsgs,
300 : 5113 : invals->msgs);
301 : 5146 : ReorderBufferXidSetCatalogChanges(ctx->reorder, xid,
302 : : buf->origptr);
303 : : }
432 michael@paquier.xyz 304 [ + - ]:GBC 9 : else if (!ctx->fast_forward)
1871 akapila@postgresql.o 305 : 9 : ReorderBufferImmediateInvalidation(ctx->reorder,
306 : 9 : invals->nmsgs,
307 : 9 : invals->msgs);
308 : :
432 michael@paquier.xyz 309 :CBC 5155 : break;
310 : : }
4205 rhaas@postgresql.org 311 : 160 : case XLOG_XACT_PREPARE:
312 : : {
313 : : xl_xact_parsed_prepare parsed;
314 : : xl_xact_prepare *xlrec;
315 : :
316 : : /* ok, parse it */
1706 akapila@postgresql.o 317 : 160 : xlrec = (xl_xact_prepare *) XLogRecGetData(r);
318 : 160 : 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 : : */
1621 326 [ + + ]: 160 : if (FilterPrepare(ctx, parsed.twophase_xid,
327 : : parsed.twophase_gid))
328 : : {
1706 329 : 17 : ReorderBufferProcessXid(reorder, parsed.twophase_xid,
330 : : buf->origptr);
331 : 17 : 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 : 143 : DecodePrepare(ctx, buf, &parsed);
348 : 143 : break;
349 : : }
4205 rhaas@postgresql.org 350 :UBC 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
1326 jdavis@postgresql.or 359 :CBC 3795 : standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
360 : : {
4205 rhaas@postgresql.org 361 : 3795 : SnapBuild *builder = ctx->snapshot_builder;
3943 heikki.linnakangas@i 362 : 3795 : XLogReaderState *r = buf->record;
363 : 3795 : uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
364 : :
3472 andres@anarazel.de 365 : 3795 : ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
366 : :
4205 rhaas@postgresql.org 367 [ + + + - ]: 3795 : switch (info)
368 : : {
369 : 1419 : case XLOG_RUNNING_XACTS:
370 : : {
3943 heikki.linnakangas@i 371 : 1419 : xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r);
372 : :
4205 rhaas@postgresql.org 373 : 1419 : 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 : 1417 : ReorderBufferAbortOld(ctx->reorder, running->oldestRunningXid);
385 : : }
386 : 1417 : break;
387 : 2367 : case XLOG_STANDBY_LOCK:
388 : 2367 : break;
3423 andres@anarazel.de 389 :GBC 9 : 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 : 9 : break;
4205 rhaas@postgresql.org 396 :UBC 0 : default:
397 [ # # ]: 0 : elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
398 : : }
4205 rhaas@postgresql.org 399 :CBC 3793 : }
400 : :
401 : : /*
402 : : * Handle rmgr HEAP2_ID records for LogicalDecodingProcessRecord().
403 : : */
404 : : void
1326 jdavis@postgresql.or 405 : 32083 : heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
406 : : {
3943 heikki.linnakangas@i 407 : 32083 : uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
408 : 32083 : TransactionId xid = XLogRecGetXid(buf->record);
4205 rhaas@postgresql.org 409 : 32083 : SnapBuild *builder = ctx->snapshot_builder;
410 : :
3472 andres@anarazel.de 411 : 32083 : 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 data changes. However, it's crucial to build the base
416 : : * snapshot during fast-forward mode (as is done in
417 : : * SnapBuildProcessChange()) because we require the snapshot's xmin when
418 : : * determining the candidate catalog_xmin for the replication slot. See
419 : : * SnapBuildProcessRunningXacts().
420 : : */
131 akapila@postgresql.o 421 [ + + ]: 32083 : if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
4205 rhaas@postgresql.org 422 : 9 : return;
423 : :
424 [ + + + + : 32074 : switch (info)
- ]
425 : : {
426 : 5901 : case XLOG_HEAP2_MULTI_INSERT:
131 akapila@postgresql.o 427 [ + - ]: 5901 : if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
428 [ + + ]: 5901 : !ctx->fast_forward)
4205 rhaas@postgresql.org 429 : 5838 : DecodeMultiInsert(ctx, buf);
430 : 5901 : break;
431 : 24403 : case XLOG_HEAP2_NEW_CID:
131 akapila@postgresql.o 432 [ + + ]: 24403 : if (!ctx->fast_forward)
433 : : {
434 : : xl_heap_new_cid *xlrec;
435 : :
3943 heikki.linnakangas@i 436 : 24222 : xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
4205 rhaas@postgresql.org 437 : 24222 : SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
438 : :
439 : 24222 : break;
440 : : }
441 : : case XLOG_HEAP2_REWRITE:
442 : :
443 : : /*
444 : : * Although these records only exist to serve the needs of logical
445 : : * decoding, all the work happens as part of crash or archive
446 : : * recovery, so we don't need to do anything here.
447 : : */
448 : 271 : break;
449 : :
450 : : /*
451 : : * Everything else here is just low level physical stuff we're not
452 : : * interested in.
453 : : */
530 heikki.linnakangas@i 454 : 1680 : case XLOG_HEAP2_PRUNE_ON_ACCESS:
455 : : case XLOG_HEAP2_PRUNE_VACUUM_SCAN:
456 : : case XLOG_HEAP2_PRUNE_VACUUM_CLEANUP:
457 : : case XLOG_HEAP2_VISIBLE:
458 : : case XLOG_HEAP2_LOCK_UPDATED:
4205 rhaas@postgresql.org 459 : 1680 : break;
4205 rhaas@postgresql.org 460 :UBC 0 : default:
461 [ # # ]: 0 : elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
462 : : }
463 : : }
464 : :
465 : : /*
466 : : * Handle rmgr HEAP_ID records for LogicalDecodingProcessRecord().
467 : : */
468 : : void
1326 jdavis@postgresql.or 469 :CBC 1412128 : heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
470 : : {
3943 heikki.linnakangas@i 471 : 1412128 : uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
472 : 1412128 : TransactionId xid = XLogRecGetXid(buf->record);
4205 rhaas@postgresql.org 473 : 1412128 : SnapBuild *builder = ctx->snapshot_builder;
474 : :
3472 andres@anarazel.de 475 : 1412128 : ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
476 : :
477 : : /*
478 : : * If we don't have snapshot or we are just fast-forwarding, there is no
479 : : * point in decoding data changes. However, it's crucial to build the base
480 : : * snapshot during fast-forward mode (as is done in
481 : : * SnapBuildProcessChange()) because we require the snapshot's xmin when
482 : : * determining the candidate catalog_xmin for the replication slot. See
483 : : * SnapBuildProcessRunningXacts().
484 : : */
131 akapila@postgresql.o 485 [ + + ]: 1412128 : if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
4205 rhaas@postgresql.org 486 : 354 : return;
487 : :
488 [ + + + + : 1411774 : switch (info)
+ + + - ]
489 : : {
490 : 870014 : case XLOG_HEAP_INSERT:
131 akapila@postgresql.o 491 [ + - ]: 870014 : if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
492 [ + + ]: 870014 : !ctx->fast_forward)
4205 rhaas@postgresql.org 493 : 869879 : DecodeInsert(ctx, buf);
494 : 870014 : break;
495 : :
496 : : /*
497 : : * Treat HOT update as normal updates. There is no useful
498 : : * information in the fact that we could make it a HOT update
499 : : * locally and the WAL layout is compatible.
500 : : */
501 : 159883 : case XLOG_HEAP_HOT_UPDATE:
502 : : case XLOG_HEAP_UPDATE:
131 akapila@postgresql.o 503 [ + - ]: 159883 : if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
504 [ + + ]: 159883 : !ctx->fast_forward)
4205 rhaas@postgresql.org 505 : 159876 : DecodeUpdate(ctx, buf);
506 : 159883 : break;
507 : :
508 : 202635 : case XLOG_HEAP_DELETE:
131 akapila@postgresql.o 509 [ + - ]: 202635 : if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
510 [ + + ]: 202635 : !ctx->fast_forward)
4205 rhaas@postgresql.org 511 : 202632 : DecodeDelete(ctx, buf);
512 : 202635 : break;
513 : :
2709 peter_e@gmx.net 514 : 54 : case XLOG_HEAP_TRUNCATE:
131 akapila@postgresql.o 515 [ + - ]: 54 : if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
516 [ + + ]: 54 : !ctx->fast_forward)
2709 peter_e@gmx.net 517 : 53 : DecodeTruncate(ctx, buf);
518 : 54 : break;
519 : :
4205 rhaas@postgresql.org 520 : 1028 : case XLOG_HEAP_INPLACE:
521 : :
522 : : /*
523 : : * Inplace updates are only ever performed on catalog tuples and
524 : : * can, per definition, not change tuple visibility. Inplace
525 : : * updates don't affect storage or interpretation of table rows,
526 : : * so they don't affect logicalrep_write_tuple() outcomes. Hence,
527 : : * we don't process invalidations from the original operation. If
528 : : * inplace updates did affect those things, invalidations wouldn't
529 : : * make it work, since there are no snapshot-specific versions of
530 : : * inplace-updated values. Since we also don't decode catalog
531 : : * tuples, we're not interested in the record's contents.
532 : : *
533 : : * WAL contains likely-unnecessary commit-time invals from the
534 : : * CacheInvalidateHeapTuple() call in
535 : : * heap_inplace_update_and_unlock(). Excess invalidation is safe.
536 : : */
537 : 1028 : break;
538 : :
3774 andres@anarazel.de 539 : 17916 : case XLOG_HEAP_CONFIRM:
131 akapila@postgresql.o 540 [ + - ]: 17916 : if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
541 [ + - ]: 17916 : !ctx->fast_forward)
3774 andres@anarazel.de 542 : 17916 : DecodeSpecConfirm(ctx, buf);
543 : 17916 : break;
544 : :
4205 rhaas@postgresql.org 545 : 160244 : case XLOG_HEAP_LOCK:
546 : : /* we don't care about row level locks for now */
547 : 160244 : break;
548 : :
4205 rhaas@postgresql.org 549 :UBC 0 : default:
550 [ # # ]: 0 : elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
551 : : break;
552 : : }
553 : : }
554 : :
555 : : /*
556 : : * Ask output plugin whether we want to skip this PREPARE and send
557 : : * this transaction as a regular commit later.
558 : : */
559 : : static inline bool
1621 akapila@postgresql.o 560 :CBC 298 : FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid,
561 : : const char *gid)
562 : : {
563 : : /*
564 : : * Skip if decoding of two-phase transactions at PREPARE time is not
565 : : * enabled. In that case, all two-phase transactions are considered
566 : : * filtered out and will be applied as regular transactions at COMMIT
567 : : * PREPARED.
568 : : */
1706 569 [ + + ]: 298 : if (!ctx->twophase)
570 : 17 : return true;
571 : :
572 : : /*
573 : : * The filter_prepare callback is optional. When not supplied, all
574 : : * prepared transactions should go through.
575 : : */
576 [ + + ]: 281 : if (ctx->callbacks.filter_prepare_cb == NULL)
577 : 133 : return false;
578 : :
1621 579 : 148 : return filter_prepare_cb_wrapper(ctx, xid, gid);
580 : : }
581 : :
582 : : static inline bool
3433 andres@anarazel.de 583 : 1248072 : FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
584 : : {
585 [ + + ]: 1248072 : if (ctx->callbacks.filter_by_origin_cb == NULL)
586 : 25 : return false;
587 : :
588 : 1248047 : return filter_by_origin_cb_wrapper(ctx, origin_id);
589 : : }
590 : :
591 : : /*
592 : : * Handle rmgr LOGICALMSG_ID records for LogicalDecodingProcessRecord().
593 : : */
594 : : void
1326 jdavis@postgresql.or 595 : 57 : logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
596 : : {
3376 rhaas@postgresql.org 597 : 57 : SnapBuild *builder = ctx->snapshot_builder;
3440 simon@2ndQuadrant.co 598 : 57 : XLogReaderState *r = buf->record;
3376 rhaas@postgresql.org 599 : 57 : TransactionId xid = XLogRecGetXid(r);
600 : 57 : uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
601 : 57 : RepOriginId origin_id = XLogRecGetOrigin(r);
927 tomas.vondra@postgre 602 : 57 : Snapshot snapshot = NULL;
603 : : xl_logical_message *message;
604 : :
3440 simon@2ndQuadrant.co 605 [ - + ]: 57 : if (info != XLOG_LOGICAL_MESSAGE)
3440 simon@2ndQuadrant.co 606 [ # # ]:UBC 0 : elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
607 : :
3440 simon@2ndQuadrant.co 608 :CBC 57 : ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
609 : :
610 : : /* If we don't have snapshot, there is no point in decoding messages */
681 akapila@postgresql.o 611 [ - + ]: 57 : if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
3440 simon@2ndQuadrant.co 612 :UBC 0 : return;
613 : :
3440 simon@2ndQuadrant.co 614 :CBC 57 : message = (xl_logical_message *) XLogRecGetData(r);
615 : :
3433 andres@anarazel.de 616 [ + + + + ]: 112 : if (message->dbId != ctx->slot->data.database ||
617 : 55 : FilterByOrigin(ctx, origin_id))
618 : 4 : return;
619 : :
3440 simon@2ndQuadrant.co 620 [ + + ]: 53 : if (message->transactional &&
621 [ - + ]: 39 : !SnapBuildProcessChange(builder, xid, buf->origptr))
3440 simon@2ndQuadrant.co 622 :UBC 0 : return;
3440 simon@2ndQuadrant.co 623 [ + + + - ]:CBC 67 : else if (!message->transactional &&
624 [ + + ]: 28 : (SnapBuildCurrentState(builder) != SNAPBUILD_CONSISTENT ||
625 : 14 : SnapBuildXactNeedsSkip(builder, buf->origptr)))
626 : 4 : return;
627 : :
628 : : /*
629 : : * We also skip decoding in fast_forward mode. This check must be last
630 : : * because we don't want to set the processing_required flag unless we
631 : : * have a decodable message.
632 : : */
681 akapila@postgresql.o 633 [ + + ]: 49 : if (ctx->fast_forward)
634 : : {
635 : : /*
636 : : * We need to set processing_required flag to notify the message's
637 : : * existence to the caller. Usually, the flag is set when either the
638 : : * COMMIT or ABORT records are decoded, but this must be turned on
639 : : * here because the non-transactional logical message is decoded
640 : : * without waiting for these records.
641 : : */
642 [ + - ]: 2 : if (!message->transactional)
643 : 2 : ctx->processing_required = true;
644 : :
645 : 2 : return;
646 : : }
647 : :
648 : : /*
649 : : * If this is a non-transactional change, get the snapshot we're expected
650 : : * to use. We only get here when the snapshot is consistent, and the
651 : : * change is not meant to be skipped.
652 : : *
653 : : * For transactional changes we don't need a snapshot, we'll use the
654 : : * regular snapshot maintained by ReorderBuffer. We just leave it NULL.
655 : : */
927 tomas.vondra@postgre 656 [ + + ]: 47 : if (!message->transactional)
657 : 8 : snapshot = SnapBuildGetOrBuildSnapshot(builder);
658 : :
3440 simon@2ndQuadrant.co 659 : 47 : ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
660 : 47 : message->transactional,
3376 rhaas@postgresql.org 661 : 47 : message->message, /* first part of message is
662 : : * prefix */
663 : : message->message_size,
3440 simon@2ndQuadrant.co 664 : 47 : message->message + message->prefix_size);
665 : : }
666 : :
667 : : /*
668 : : * Consolidated commit record handling between the different form of commit
669 : : * records.
670 : : *
671 : : * 'two_phase' indicates that caller wants to process the transaction in two
672 : : * phases, first process prepare if not already done and then process
673 : : * commit_prepared.
674 : : */
675 : : static void
4205 rhaas@postgresql.org 676 : 3037 : DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
677 : : xl_xact_parsed_commit *parsed, TransactionId xid,
678 : : bool two_phase)
679 : : {
3783 andres@anarazel.de 680 : 3037 : XLogRecPtr origin_lsn = InvalidXLogRecPtr;
3376 rhaas@postgresql.org 681 : 3037 : TimestampTz commit_time = parsed->xact_time;
682 : 3037 : RepOriginId origin_id = XLogRecGetOrigin(buf->record);
683 : : int i;
684 : :
3783 andres@anarazel.de 685 [ + + ]: 3037 : if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
686 : : {
687 : 76 : origin_lsn = parsed->origin_lsn;
688 : 76 : commit_time = parsed->origin_timestamp;
689 : : }
690 : :
4205 rhaas@postgresql.org 691 : 3037 : SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
692 : : parsed->nsubxacts, parsed->subxacts,
693 : : parsed->xinfo);
694 : :
695 : : /* ----
696 : : * Check whether we are interested in this specific transaction, and tell
697 : : * the reorderbuffer to forget the content of the (sub-)transactions
698 : : * if not.
699 : : *
700 : : * We can't just use ReorderBufferAbort() here, because we need to execute
701 : : * the transaction's invalidations. This currently won't be needed if
702 : : * we're just skipping over the transaction because currently we only do
703 : : * so during startup, to get to the first transaction the client needs. As
704 : : * we have reset the catalog caches before starting to read WAL, and we
705 : : * haven't yet touched any catalogs, there can't be anything to invalidate.
706 : : * But if we're "forgetting" this commit because it happened in another
707 : : * database, the invalidations might be important, because they could be
708 : : * for shared catalogs and we might have loaded data into the relevant
709 : : * syscaches.
710 : : * ---
711 : : */
1706 akapila@postgresql.o 712 [ + + ]: 3037 : if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
713 : : {
3828 andres@anarazel.de 714 [ + + ]: 2575 : for (i = 0; i < parsed->nsubxacts; i++)
715 : : {
716 : 943 : ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
717 : : }
4205 rhaas@postgresql.org 718 : 1632 : ReorderBufferForget(ctx->reorder, xid, buf->origptr);
719 : :
720 : 1632 : return;
721 : : }
722 : :
723 : : /* tell the reorderbuffer about the surviving subtransactions */
3828 andres@anarazel.de 724 [ + + ]: 1671 : for (i = 0; i < parsed->nsubxacts; i++)
725 : : {
726 : 266 : ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
727 : : buf->origptr, buf->endptr);
728 : : }
729 : :
730 : : /*
731 : : * Send the final commit record if the transaction data is already
732 : : * decoded, otherwise, process the entire transaction.
733 : : */
1706 akapila@postgresql.o 734 [ + + ]: 1405 : if (two_phase)
735 : : {
736 : 31 : ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
1515 737 : 31 : SnapBuildGetTwoPhaseAt(ctx->snapshot_builder),
738 : : commit_time, origin_id, origin_lsn,
1706 739 : 31 : parsed->twophase_gid, true);
740 : : }
741 : : else
742 : : {
743 : 1374 : ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
744 : : commit_time, origin_id, origin_lsn);
745 : : }
746 : :
747 : : /*
748 : : * Update the decoding stats at transaction prepare/commit/abort.
749 : : * Additionally we send the stats when we spill or stream the changes to
750 : : * avoid losing them in case the decoding is interrupted. It is not clear
751 : : * that sending more or less frequently than this would be better.
752 : : */
753 : 1398 : UpdateDecodingStats(ctx);
754 : : }
755 : :
756 : : /*
757 : : * Decode PREPARE record. Similar logic as in DecodeCommit.
758 : : *
759 : : * Note that we don't skip prepare even if have detected concurrent abort
760 : : * because it is quite possible that we had already sent some changes before we
761 : : * detect abort in which case we need to abort those changes in the subscriber.
762 : : * To abort such changes, we do send the prepare and then the rollback prepared
763 : : * which is what happened on the publisher-side as well. Now, we can invent a
764 : : * new abort API wherein in such cases we send abort and skip sending prepared
765 : : * and rollback prepared but then it is not that straightforward because we
766 : : * might have streamed this transaction by that time in which case it is
767 : : * handled when the rollback is encountered. It is not impossible to optimize
768 : : * the concurrent abort case but it can introduce design complexity w.r.t
769 : : * handling different cases so leaving it for now as it doesn't seem worth it.
770 : : */
771 : : static void
772 : 143 : DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
773 : : xl_xact_parsed_prepare *parsed)
774 : : {
775 : 143 : SnapBuild *builder = ctx->snapshot_builder;
776 : 143 : XLogRecPtr origin_lsn = parsed->origin_lsn;
777 : 143 : TimestampTz prepare_time = parsed->xact_time;
841 tgl@sss.pgh.pa.us 778 : 143 : RepOriginId origin_id = XLogRecGetOrigin(buf->record);
779 : : int i;
1706 akapila@postgresql.o 780 : 143 : TransactionId xid = parsed->twophase_xid;
781 : :
782 [ + + ]: 143 : if (parsed->origin_timestamp != 0)
783 : 8 : prepare_time = parsed->origin_timestamp;
784 : :
785 : : /*
786 : : * Remember the prepare info for a txn so that it can be used later in
787 : : * commit prepared if required. See ReorderBufferFinishPrepared.
788 : : */
789 [ - + ]: 143 : if (!ReorderBufferRememberPrepareInfo(ctx->reorder, xid, buf->origptr,
790 : : buf->endptr, prepare_time, origin_id,
791 : : origin_lsn))
1706 akapila@postgresql.o 792 :UBC 0 : return;
793 : :
794 : : /* We can't start streaming unless a consistent state is reached. */
1706 akapila@postgresql.o 795 [ + + ]:CBC 143 : if (SnapBuildCurrentState(builder) < SNAPBUILD_CONSISTENT)
796 : : {
797 : 3 : ReorderBufferSkipPrepare(ctx->reorder, xid);
798 : 3 : return;
799 : : }
800 : :
801 : : /*
802 : : * Check whether we need to process this transaction. See
803 : : * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
804 : : * transaction.
805 : : *
806 : : * We can't call ReorderBufferForget as we did in DecodeCommit as the txn
807 : : * hasn't yet been committed, removing this txn before a commit might
808 : : * result in the computation of an incorrect restart_lsn. See
809 : : * SnapBuildProcessRunningXacts. But we need to process cache
810 : : * invalidations if there are any for the reasons mentioned in
811 : : * DecodeCommit.
812 : : */
813 [ + + ]: 140 : if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
814 : : {
815 : 101 : ReorderBufferSkipPrepare(ctx->reorder, xid);
816 : 101 : ReorderBufferInvalidate(ctx->reorder, xid, buf->origptr);
817 : 101 : return;
818 : : }
819 : :
820 : : /* Tell the reorderbuffer about the surviving subtransactions. */
821 [ + + ]: 40 : for (i = 0; i < parsed->nsubxacts; i++)
822 : : {
823 : 1 : ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
824 : : buf->origptr, buf->endptr);
825 : : }
826 : :
827 : : /* replay actions of all transaction + subtransactions in order */
828 : 39 : ReorderBufferPrepare(ctx->reorder, xid, parsed->twophase_gid);
829 : :
830 : : /*
831 : : * Update the decoding stats at transaction prepare/commit/abort.
832 : : * Additionally we send the stats when we spill or stream the changes to
833 : : * avoid losing them in case the decoding is interrupted. It is not clear
834 : : * that sending more or less frequently than this would be better.
835 : : */
1794 836 : 39 : UpdateDecodingStats(ctx);
837 : : }
838 : :
839 : :
840 : : /*
841 : : * Get the data from the various forms of abort records and pass it on to
842 : : * snapbuild.c and reorderbuffer.c.
843 : : *
844 : : * 'two_phase' indicates to finish prepared transaction.
845 : : */
846 : : static void
3828 andres@anarazel.de 847 : 153 : DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
848 : : xl_xact_parsed_abort *parsed, TransactionId xid,
849 : : bool two_phase)
850 : : {
851 : : int i;
1706 akapila@postgresql.o 852 : 153 : XLogRecPtr origin_lsn = InvalidXLogRecPtr;
853 : 153 : TimestampTz abort_time = parsed->xact_time;
841 tgl@sss.pgh.pa.us 854 : 153 : RepOriginId origin_id = XLogRecGetOrigin(buf->record);
855 : : bool skip_xact;
856 : :
1706 akapila@postgresql.o 857 [ + + ]: 153 : if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
858 : : {
859 : 4 : origin_lsn = parsed->origin_lsn;
860 : 4 : abort_time = parsed->origin_timestamp;
861 : : }
862 : :
863 : : /*
864 : : * Check whether we need to process this transaction. See
865 : : * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
866 : : * transaction.
867 : : */
868 : 153 : skip_xact = DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id);
869 : :
870 : : /*
871 : : * Send the final rollback record for a prepared transaction unless we
872 : : * need to skip it. For non-two-phase xacts, simply forget the xact.
873 : : */
874 [ + + + + ]: 153 : if (two_phase && !skip_xact)
875 : : {
876 : 10 : ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
877 : : InvalidXLogRecPtr,
878 : : abort_time, origin_id, origin_lsn,
879 : 10 : parsed->twophase_gid, false);
880 : : }
881 : : else
882 : : {
883 [ + + ]: 149 : for (i = 0; i < parsed->nsubxacts; i++)
884 : : {
885 : 6 : ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
971 886 : 6 : buf->record->EndRecPtr, abort_time);
887 : : }
888 : :
889 : 143 : ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr,
890 : : abort_time);
891 : : }
892 : :
893 : : /* update the decoding stats */
1794 894 : 153 : UpdateDecodingStats(ctx);
4205 rhaas@postgresql.org 895 : 153 : }
896 : :
897 : : /*
898 : : * Parse XLOG_HEAP_INSERT (not MULTI_INSERT!) records into tuplebufs.
899 : : *
900 : : * Inserts can contain the new tuple.
901 : : */
902 : : static void
903 : 869879 : DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
904 : : {
905 : : Size datalen;
906 : : char *tupledata;
907 : : Size tuplelen;
3943 heikki.linnakangas@i 908 : 869879 : XLogReaderState *r = buf->record;
909 : : xl_heap_insert *xlrec;
910 : : ReorderBufferChange *change;
911 : : RelFileLocator target_locator;
912 : :
913 : 869879 : xlrec = (xl_heap_insert *) XLogRecGetData(r);
914 : :
915 : : /*
916 : : * Ignore insert records without new tuples (this does happen when
917 : : * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
918 : : */
2474 tomas.vondra@postgre 919 [ + + ]: 869879 : if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
920 : 3885 : return;
921 : :
922 : : /* only interested in our database */
1158 rhaas@postgresql.org 923 : 866056 : XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
924 [ - + ]: 866056 : if (target_locator.dbOid != ctx->slot->data.database)
4205 rhaas@postgresql.org 925 :UBC 0 : return;
926 : :
927 : : /* output plugin doesn't look for this origin, no need to queue */
3783 andres@anarazel.de 928 [ + + ]:CBC 866056 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
929 : 62 : return;
930 : :
178 heikki.linnakangas@i 931 : 865994 : change = ReorderBufferAllocChange(ctx->reorder);
3774 andres@anarazel.de 932 [ + + ]: 865994 : if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
933 : 848078 : change->action = REORDER_BUFFER_CHANGE_INSERT;
934 : : else
935 : 17916 : change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT;
3783 936 : 865994 : change->origin_id = XLogRecGetOrigin(r);
937 : :
1158 rhaas@postgresql.org 938 : 865994 : memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
939 : :
2474 tomas.vondra@postgre 940 : 865994 : tupledata = XLogRecGetBlockData(r, 0, &datalen);
941 : 865994 : tuplelen = datalen - SizeOfHeapHeader;
942 : :
943 : 865994 : change->data.tp.newtuple =
178 heikki.linnakangas@i 944 : 865994 : ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);
945 : :
2474 tomas.vondra@postgre 946 : 865994 : DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
947 : :
4080 andres@anarazel.de 948 : 865994 : change->data.tp.clear_toast_afterwards = true;
949 : :
1855 akapila@postgresql.o 950 : 865994 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
951 : : change,
952 : 865994 : xlrec->flags & XLH_INSERT_ON_TOAST_RELATION);
953 : : }
954 : :
955 : : /*
956 : : * Parse XLOG_HEAP_UPDATE and XLOG_HEAP_HOT_UPDATE, which have the same layout
957 : : * in the record, from wal into proper tuplebufs.
958 : : *
959 : : * Updates can possibly contain a new tuple and the old primary key.
960 : : */
961 : : static void
4205 rhaas@postgresql.org 962 : 159876 : DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
963 : : {
3943 heikki.linnakangas@i 964 : 159876 : XLogReaderState *r = buf->record;
965 : : xl_heap_update *xlrec;
966 : : ReorderBufferChange *change;
967 : : char *data;
968 : : RelFileLocator target_locator;
969 : :
970 : 159876 : xlrec = (xl_heap_update *) XLogRecGetData(r);
971 : :
972 : : /* only interested in our database */
1158 rhaas@postgresql.org 973 : 159876 : XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
974 [ + + ]: 159876 : if (target_locator.dbOid != ctx->slot->data.database)
4205 975 : 92 : return;
976 : :
977 : : /* output plugin doesn't look for this origin, no need to queue */
3783 andres@anarazel.de 978 [ + + ]: 159812 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
979 : 28 : return;
980 : :
178 heikki.linnakangas@i 981 : 159784 : change = ReorderBufferAllocChange(ctx->reorder);
4205 rhaas@postgresql.org 982 : 159784 : change->action = REORDER_BUFFER_CHANGE_UPDATE;
3783 andres@anarazel.de 983 : 159784 : change->origin_id = XLogRecGetOrigin(r);
1158 rhaas@postgresql.org 984 : 159784 : memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
985 : :
3774 andres@anarazel.de 986 [ + + ]: 159784 : if (xlrec->flags & XLH_UPDATE_CONTAINS_NEW_TUPLE)
987 : : {
988 : : Size datalen;
989 : : Size tuplelen;
990 : :
3943 heikki.linnakangas@i 991 : 158037 : data = XLogRecGetBlockData(r, 0, &datalen);
992 : :
3470 andres@anarazel.de 993 : 158037 : tuplelen = datalen - SizeOfHeapHeader;
994 : :
3472 995 : 158037 : change->data.tp.newtuple =
178 heikki.linnakangas@i 996 : 158037 : ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);
997 : :
3943 998 : 158037 : DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
999 : : }
1000 : :
3774 andres@anarazel.de 1001 [ + + ]: 159784 : if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
1002 : : {
1003 : : Size datalen;
1004 : : Size tuplelen;
1005 : :
1006 : : /* caution, remaining data in record is not aligned */
3943 heikki.linnakangas@i 1007 : 396 : data = XLogRecGetData(r) + SizeOfHeapUpdate;
1008 : 396 : datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
3470 andres@anarazel.de 1009 : 396 : tuplelen = datalen - SizeOfHeapHeader;
1010 : :
3472 1011 : 396 : change->data.tp.oldtuple =
178 heikki.linnakangas@i 1012 : 396 : ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);
1013 : :
3943 1014 : 396 : DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
1015 : : }
1016 : :
4080 andres@anarazel.de 1017 : 159784 : change->data.tp.clear_toast_afterwards = true;
1018 : :
1855 akapila@postgresql.o 1019 : 159784 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
1020 : : change, false);
1021 : : }
1022 : :
1023 : : /*
1024 : : * Parse XLOG_HEAP_DELETE from wal into proper tuplebufs.
1025 : : *
1026 : : * Deletes can possibly contain the old primary key.
1027 : : */
1028 : : static void
4205 rhaas@postgresql.org 1029 : 202632 : DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
1030 : : {
3943 heikki.linnakangas@i 1031 : 202632 : XLogReaderState *r = buf->record;
1032 : : xl_heap_delete *xlrec;
1033 : : ReorderBufferChange *change;
1034 : : RelFileLocator target_locator;
1035 : :
1036 : 202632 : xlrec = (xl_heap_delete *) XLogRecGetData(r);
1037 : :
1038 : : /* only interested in our database */
1158 rhaas@postgresql.org 1039 : 202632 : XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
1040 [ + + ]: 202632 : if (target_locator.dbOid != ctx->slot->data.database)
4205 1041 : 38 : return;
1042 : :
1043 : : /* output plugin doesn't look for this origin, no need to queue */
3783 andres@anarazel.de 1044 [ + + ]: 202605 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1045 : 11 : return;
1046 : :
178 heikki.linnakangas@i 1047 : 202594 : change = ReorderBufferAllocChange(ctx->reorder);
1048 : :
1544 akapila@postgresql.o 1049 [ - + ]: 202594 : if (xlrec->flags & XLH_DELETE_IS_SUPER)
1544 akapila@postgresql.o 1050 :UBC 0 : change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT;
1051 : : else
1544 akapila@postgresql.o 1052 :CBC 202594 : change->action = REORDER_BUFFER_CHANGE_DELETE;
1053 : :
3783 andres@anarazel.de 1054 : 202594 : change->origin_id = XLogRecGetOrigin(r);
1055 : :
1158 rhaas@postgresql.org 1056 : 202594 : memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
1057 : :
1058 : : /* old primary key stored */
3774 andres@anarazel.de 1059 [ + + ]: 202594 : if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
1060 : : {
3470 1061 : 140769 : Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
1062 : 140769 : Size tuplelen = datalen - SizeOfHeapHeader;
1063 : :
3943 heikki.linnakangas@i 1064 [ - + ]: 140769 : Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
1065 : :
3472 andres@anarazel.de 1066 : 140769 : change->data.tp.oldtuple =
178 heikki.linnakangas@i 1067 : 140769 : ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);
1068 : :
4205 rhaas@postgresql.org 1069 : 140769 : DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
1070 : : datalen, change->data.tp.oldtuple);
1071 : : }
1072 : :
4080 andres@anarazel.de 1073 : 202594 : change->data.tp.clear_toast_afterwards = true;
1074 : :
1855 akapila@postgresql.o 1075 : 202594 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
1076 : : change, false);
1077 : : }
1078 : :
1079 : : /*
1080 : : * Parse XLOG_HEAP_TRUNCATE from wal
1081 : : */
1082 : : static void
2709 peter_e@gmx.net 1083 : 53 : DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
1084 : : {
1085 : 53 : XLogReaderState *r = buf->record;
1086 : : xl_heap_truncate *xlrec;
1087 : : ReorderBufferChange *change;
1088 : :
1089 : 53 : xlrec = (xl_heap_truncate *) XLogRecGetData(r);
1090 : :
1091 : : /* only interested in our database */
1092 [ - + ]: 53 : if (xlrec->dbId != ctx->slot->data.database)
2709 peter_e@gmx.net 1093 :UBC 0 : return;
1094 : :
1095 : : /* output plugin doesn't look for this origin, no need to queue */
2709 peter_e@gmx.net 1096 [ - + ]:CBC 53 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
2709 peter_e@gmx.net 1097 :UBC 0 : return;
1098 : :
178 heikki.linnakangas@i 1099 :CBC 53 : change = ReorderBufferAllocChange(ctx->reorder);
2709 peter_e@gmx.net 1100 : 53 : change->action = REORDER_BUFFER_CHANGE_TRUNCATE;
1101 : 53 : change->origin_id = XLogRecGetOrigin(r);
1102 [ + + ]: 53 : if (xlrec->flags & XLH_TRUNCATE_CASCADE)
1103 : 1 : change->data.truncate.cascade = true;
1104 [ + + ]: 53 : if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
1105 : 2 : change->data.truncate.restart_seqs = true;
1106 : 53 : change->data.truncate.nrelids = xlrec->nrelids;
178 heikki.linnakangas@i 1107 : 106 : change->data.truncate.relids = ReorderBufferAllocRelids(ctx->reorder,
1108 : 53 : xlrec->nrelids);
2709 peter_e@gmx.net 1109 : 53 : memcpy(change->data.truncate.relids, xlrec->relids,
1110 : 53 : xlrec->nrelids * sizeof(Oid));
1111 : 53 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
1112 : : buf->origptr, change, false);
1113 : : }
1114 : :
1115 : : /*
1116 : : * Decode XLOG_HEAP2_MULTI_INSERT record into multiple tuplebufs.
1117 : : *
1118 : : * Currently MULTI_INSERT will always contain the full tuples.
1119 : : */
1120 : : static void
4205 rhaas@postgresql.org 1121 : 5838 : DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
1122 : : {
3943 heikki.linnakangas@i 1123 : 5838 : XLogReaderState *r = buf->record;
1124 : : xl_heap_multi_insert *xlrec;
1125 : : int i;
1126 : : char *data;
1127 : : char *tupledata;
1128 : : Size tuplelen;
1129 : : RelFileLocator rlocator;
1130 : :
1131 : 5838 : xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
1132 : :
1133 : : /*
1134 : : * Ignore insert records without new tuples. This happens when a
1135 : : * multi_insert is done on a catalog or on a non-persistent relation.
1136 : : */
2014 michael@paquier.xyz 1137 [ + + ]: 5838 : if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
1138 : 5823 : return;
1139 : :
1140 : : /* only interested in our database */
1158 rhaas@postgresql.org 1141 : 66 : XLogRecGetBlockTag(r, 0, &rlocator, NULL, NULL);
1142 [ + + ]: 66 : if (rlocator.dbOid != ctx->slot->data.database)
4205 1143 : 51 : return;
1144 : :
1145 : : /* output plugin doesn't look for this origin, no need to queue */
3783 andres@anarazel.de 1146 [ - + ]: 15 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
3783 andres@anarazel.de 1147 :UBC 0 : return;
1148 : :
1149 : : /*
1150 : : * We know that this multi_insert isn't for a catalog, so the block should
1151 : : * always have data even if a full-page write of it is taken.
1152 : : */
3943 heikki.linnakangas@i 1153 :CBC 15 : tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
2222 michael@paquier.xyz 1154 [ - + ]: 15 : Assert(tupledata != NULL);
1155 : :
3943 heikki.linnakangas@i 1156 : 15 : data = tupledata;
4205 rhaas@postgresql.org 1157 [ + + ]: 1064 : for (i = 0; i < xlrec->ntuples; i++)
1158 : : {
1159 : : ReorderBufferChange *change;
1160 : : xl_multi_insert_tuple *xlhdr;
1161 : : int datalen;
1162 : : HeapTuple tuple;
1163 : : HeapTupleHeader header;
1164 : :
178 heikki.linnakangas@i 1165 : 1049 : change = ReorderBufferAllocChange(ctx->reorder);
4205 rhaas@postgresql.org 1166 : 1049 : change->action = REORDER_BUFFER_CHANGE_INSERT;
3783 andres@anarazel.de 1167 : 1049 : change->origin_id = XLogRecGetOrigin(r);
1168 : :
1158 rhaas@postgresql.org 1169 : 1049 : memcpy(&change->data.tp.rlocator, &rlocator, sizeof(RelFileLocator));
1170 : :
2222 michael@paquier.xyz 1171 : 1049 : xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
1172 : 1049 : data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
1173 : 1049 : datalen = xlhdr->datalen;
1174 : :
2014 1175 : 1049 : change->data.tp.newtuple =
178 heikki.linnakangas@i 1176 : 1049 : ReorderBufferAllocTupleBuf(ctx->reorder, datalen);
1177 : :
2014 michael@paquier.xyz 1178 : 1049 : tuple = change->data.tp.newtuple;
586 msawada@postgresql.o 1179 : 1049 : header = tuple->t_data;
1180 : :
1181 : : /* not a disk based tuple */
1182 : 1049 : ItemPointerSetInvalid(&tuple->t_self);
1183 : :
1184 : : /*
1185 : : * We can only figure this out after reassembling the transactions.
1186 : : */
1187 : 1049 : tuple->t_tableOid = InvalidOid;
1188 : :
1189 : 1049 : tuple->t_len = datalen + SizeofHeapTupleHeader;
1190 : :
2014 michael@paquier.xyz 1191 : 1049 : memset(header, 0, SizeofHeapTupleHeader);
1192 : :
206 peter@eisentraut.org 1193 : 1049 : memcpy((char *) tuple->t_data + SizeofHeapTupleHeader, data, datalen);
2014 michael@paquier.xyz 1194 : 1049 : header->t_infomask = xlhdr->t_infomask;
1195 : 1049 : header->t_infomask2 = xlhdr->t_infomask2;
1196 : 1049 : header->t_hoff = xlhdr->t_hoff;
1197 : :
1198 : : /*
1199 : : * Reset toast reassembly state only after the last row in the last
1200 : : * xl_multi_insert_tuple record emitted by one heap_multi_insert()
1201 : : * call.
1202 : : */
3774 andres@anarazel.de 1203 [ + + ]: 1049 : if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
4074 1204 [ + + ]: 189 : (i + 1) == xlrec->ntuples)
1205 : 10 : change->data.tp.clear_toast_afterwards = true;
1206 : : else
1207 : 1039 : change->data.tp.clear_toast_afterwards = false;
1208 : :
3943 heikki.linnakangas@i 1209 : 1049 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
1210 : : buf->origptr, change, false);
1211 : :
1212 : : /* move to the next xl_multi_insert_tuple entry */
2222 michael@paquier.xyz 1213 : 1049 : data += datalen;
1214 : : }
3943 heikki.linnakangas@i 1215 [ - + ]: 15 : Assert(data == tupledata + tuplelen);
1216 : : }
1217 : :
1218 : : /*
1219 : : * Parse XLOG_HEAP_CONFIRM from wal into a confirmation change.
1220 : : *
1221 : : * This is pretty trivial, all the state essentially already setup by the
1222 : : * speculative insertion.
1223 : : */
1224 : : static void
3774 andres@anarazel.de 1225 : 17916 : DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
1226 : : {
1227 : 17916 : XLogReaderState *r = buf->record;
1228 : : ReorderBufferChange *change;
1229 : : RelFileLocator target_locator;
1230 : :
1231 : : /* only interested in our database */
1158 rhaas@postgresql.org 1232 : 17916 : XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
1233 [ - + ]: 17916 : if (target_locator.dbOid != ctx->slot->data.database)
3774 andres@anarazel.de 1234 :UBC 0 : return;
1235 : :
1236 : : /* output plugin doesn't look for this origin, no need to queue */
3774 andres@anarazel.de 1237 [ - + ]:CBC 17916 : if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
3774 andres@anarazel.de 1238 :UBC 0 : return;
1239 : :
178 heikki.linnakangas@i 1240 :CBC 17916 : change = ReorderBufferAllocChange(ctx->reorder);
3774 andres@anarazel.de 1241 : 17916 : change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM;
1242 : 17916 : change->origin_id = XLogRecGetOrigin(r);
1243 : :
1158 rhaas@postgresql.org 1244 : 17916 : memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
1245 : :
3774 andres@anarazel.de 1246 : 17916 : change->data.tp.clear_toast_afterwards = true;
1247 : :
1855 akapila@postgresql.o 1248 : 17916 : ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
1249 : : change, false);
1250 : : }
1251 : :
1252 : :
1253 : : /*
1254 : : * Read a HeapTuple as WAL logged by heap_insert, heap_update and heap_delete
1255 : : * (but not by heap_multi_insert) into a tuplebuf.
1256 : : *
1257 : : * The size 'len' and the pointer 'data' in the record need to be
1258 : : * computed outside as they are record specific.
1259 : : */
1260 : : static void
586 msawada@postgresql.o 1261 : 1165196 : DecodeXLogTuple(char *data, Size len, HeapTuple tuple)
1262 : : {
1263 : : xl_heap_header xlhdr;
4205 rhaas@postgresql.org 1264 : 1165196 : int datalen = len - SizeOfHeapHeader;
1265 : : HeapTupleHeader header;
1266 : :
1267 [ - + ]: 1165196 : Assert(datalen >= 0);
1268 : :
586 msawada@postgresql.o 1269 : 1165196 : tuple->t_len = datalen + SizeofHeapTupleHeader;
1270 : 1165196 : header = tuple->t_data;
1271 : :
1272 : : /* not a disk based tuple */
1273 : 1165196 : ItemPointerSetInvalid(&tuple->t_self);
1274 : :
1275 : : /* we can only figure this out after reassembling the transactions */
1276 : 1165196 : tuple->t_tableOid = InvalidOid;
1277 : :
1278 : : /* data is not stored aligned, copy to aligned storage */
206 peter@eisentraut.org 1279 : 1165196 : memcpy(&xlhdr, data, SizeOfHeapHeader);
1280 : :
3472 andres@anarazel.de 1281 : 1165196 : memset(header, 0, SizeofHeapTupleHeader);
1282 : :
586 msawada@postgresql.o 1283 : 1165196 : memcpy(((char *) tuple->t_data) + SizeofHeapTupleHeader,
4205 rhaas@postgresql.org 1284 : 1165196 : data + SizeOfHeapHeader,
1285 : : datalen);
1286 : :
3472 andres@anarazel.de 1287 : 1165196 : header->t_infomask = xlhdr.t_infomask;
1288 : 1165196 : header->t_infomask2 = xlhdr.t_infomask2;
1289 : 1165196 : header->t_hoff = xlhdr.t_hoff;
4205 rhaas@postgresql.org 1290 : 1165196 : }
1291 : :
1292 : : /*
1293 : : * Check whether we are interested in this specific transaction.
1294 : : *
1295 : : * There can be several reasons we might not be interested in this
1296 : : * transaction:
1297 : : * 1) We might not be interested in decoding transactions up to this
1298 : : * LSN. This can happen because we previously decoded it and now just
1299 : : * are restarting or if we haven't assembled a consistent snapshot yet.
1300 : : * 2) The transaction happened in another database.
1301 : : * 3) The output plugin is not interested in the origin.
1302 : : * 4) We are doing fast-forwarding
1303 : : */
1304 : : static bool
1706 akapila@postgresql.o 1305 : 3330 : DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
1306 : : Oid txn_dbid, RepOriginId origin_id)
1307 : : {
681 1308 [ + + + + ]: 3330 : if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
1309 [ + + + + ]: 3091 : (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) ||
1310 : 1560 : FilterByOrigin(ctx, origin_id))
1311 : 1801 : return true;
1312 : :
1313 : : /*
1314 : : * We also skip decoding in fast_forward mode. In passing set the
1315 : : * processing_required flag to indicate that if it were not for
1316 : : * fast_forward mode, processing would have been required.
1317 : : */
1318 [ + + ]: 1529 : if (ctx->fast_forward)
1319 : : {
1320 : 23 : ctx->processing_required = true;
1321 : 23 : return true;
1322 : : }
1323 : :
1324 : 1506 : return false;
1325 : : }
|
