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