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