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