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