Loading...
Searching...
No Matches
reorderbuffer.c File Reference
#include "postgres.h"#include <unistd.h>#include <sys/stat.h>#include "access/detoast.h"#include "access/heapam.h"#include "access/rewriteheap.h"#include "access/transam.h"#include "access/xact.h"#include "access/xlog_internal.h"#include "catalog/catalog.h"#include "common/int.h"#include "lib/binaryheap.h"#include "miscadmin.h"#include "pgstat.h"#include "replication/logical.h"#include "replication/reorderbuffer.h"#include "replication/slot.h"#include "replication/snapbuild.h"#include "storage/bufmgr.h"#include "storage/fd.h"#include "storage/procarray.h"#include "storage/sinval.h"#include "utils/builtins.h"#include "utils/inval.h"#include "utils/memutils.h"#include "utils/rel.h"#include "utils/relfilenumbermap.h"#include "utils/wait_event.h"
Include dependency graph for reorderbuffer.c:
Go to the source code of this file.
Data Structures | |
| struct | ReorderBufferTXNByIdEnt |
| struct | ReorderBufferTupleCidKey |
| struct | ReorderBufferTupleCidEnt |
| struct | TXNEntryFile |
| struct | ReorderBufferIterTXNEntry |
| struct | ReorderBufferIterTXNState |
| struct | ReorderBufferToastEnt |
| struct | ReorderBufferDiskChange |
| struct | RewriteMappingFile |
Macros | |
| #define | MAX_DISTR_INVAL_MSG_PER_TXN ((8 * 1024 * 1024) / sizeof(SharedInvalidationMessage)) |
| #define | IsSpecInsert(action) |
| #define | IsSpecConfirmOrAbort(action) |
| #define | IsInsertOrUpdate(action) |
| #define | CHANGES_THRESHOLD 100 |
Variables | |
| int | logical_decoding_work_mem |
| static const Size | max_changes_in_memory = 4096 |
| int | debug_logical_replication_streaming = DEBUG_LOGICAL_REP_STREAMING_BUFFERED |
Macro Definition Documentation
◆ CHANGES_THRESHOLD
| #define CHANGES_THRESHOLD 100 |
◆ IsInsertOrUpdate
| #define IsInsertOrUpdate | ( | action | ) |
Value:
( \
(((action) == REORDER_BUFFER_CHANGE_INSERT) || \
((action) == REORDER_BUFFER_CHANGE_UPDATE) || \
((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT)) \
)
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT
Definition reorderbuffer.h:60
Definition at line 207 of file reorderbuffer.c.
325{
326 ReorderBuffer *buffer;
329
331
332 /* allocate memory in own context, to have better accountability */
334 "ReorderBuffer",
335 ALLOCSET_DEFAULT_SIZES);
336
337 buffer =
339
341
343
345 "Change",
348
350 "TXN",
353
354 /*
355 * To minimize memory fragmentation caused by long-running transactions
356 * with changes spanning multiple memory blocks, we use a single
357 * fixed-size memory block for decoded tuple storage. The performance
358 * testing showed that the default memory block size maintains logical
359 * decoding performance without causing fragmentation due to concurrent
360 * transactions. One might think that we can use the max size as
361 * SLAB_LARGE_BLOCK_SIZE but the test also showed it doesn't help resolve
362 * the memory fragmentation.
363 */
365 "Tuples",
368 SLAB_DEFAULT_BLOCK_SIZE);
369
373
376
379
381 buffer->outbufsize = 0;
382 buffer->size = 0;
383
384 /* txn_heap is ordered by transaction size */
386
387 buffer->spillTxns = 0;
388 buffer->spillCount = 0;
389 buffer->spillBytes = 0;
390 buffer->streamTxns = 0;
391 buffer->streamCount = 0;
392 buffer->streamBytes = 0;
393 buffer->memExceededCount = 0;
394 buffer->totalTxns = 0;
395 buffer->totalBytes = 0;
396
398
402
403 /*
404 * Ensure there's no stale data from prior uses of this slot, in case some
405 * prior exit avoided calling ReorderBufferFree. Failure to do this can
406 * produce duplicated txns, and it's very cheap if there's nothing there.
407 */
409
410 return buffer;
411}
412
413/*
414 * Free a ReorderBuffer
415 */
416void
418{
420
421 /*
422 * We free separately allocated data by entirely scrapping reorderbuffer's
423 * memory context.
424 */
425 MemoryContextDelete(context);
426
427 /* Free disk space used by unconsumed reorder buffers */
429}
430
431/*
432 * Allocate a new ReorderBufferTXN.
433 */
436{
437 ReorderBufferTXN *txn;
438
439 txn = (ReorderBufferTXN *)
441
443
447
448 /* InvalidCommandId is not zero, so set it explicitly */
451
452 return txn;
453}
454
455/*
456 * Free a ReorderBufferTXN.
457 */
458static void
460{
461 /* clean the lookup cache if we were cached (quite likely) */
463 {
466 }
467
468 /* free data that's contained */
469
471 {
474 }
475
477 {
480 }
481
483 {
486 }
487
489 {
492 }
493
494 /* Reset the toast hash */
496
497 /* All changes must be deallocated */
499
500 pfree(txn);
501}
502
503/*
504 * Allocate a ReorderBufferChange.
505 */
506ReorderBufferChange *
508{
509 ReorderBufferChange *change;
510
511 change = (ReorderBufferChange *)
513
515 return change;
516}
517
518/*
519 * Free a ReorderBufferChange and update memory accounting, if requested.
520 */
521void
524{
525 /* update memory accounting info */
528 ReorderBufferChangeSize(change));
529
530 /* free contained data */
532 {
538 {
541 }
542
544 {
547 }
548 break;
556 break;
561 break;
564 {
567 }
568 break;
569 /* no data in addition to the struct itself */
572 {
575 }
576 break;
581 break;
582 }
583
584 pfree(change);
585}
586
587/*
588 * Allocate a HeapTuple fitting a tuple of size tuple_len (excluding header
589 * overhead).
590 */
591HeapTuple
593{
594 HeapTuple tuple;
596
598
602
603 return tuple;
604}
605
606/*
607 * Free a HeapTuple returned by ReorderBufferAllocTupleBuf().
608 */
609void
611{
612 pfree(tuple);
613}
614
615/*
616 * Allocate an array for relids of truncated relations.
617 *
618 * We use the global memory context (for the whole reorder buffer), because
619 * none of the existing ones seems like a good match (some are SLAB, so we
620 * can't use those, and tup_context is meant for tuple data, not relids). We
621 * could add yet another context, but it seems like an overkill - TRUNCATE is
622 * not particularly common operation, so it does not seem worth it.
623 */
624Oid *
626{
627 Oid *relids;
629
631
633
634 return relids;
635}
636
637/*
638 * Free an array of relids.
639 */
640void
642{
643 pfree(relids);
644}
645
646/*
647 * Return the ReorderBufferTXN from the given buffer, specified by Xid.
648 * If create is true, and a transaction doesn't already exist, create it
649 * (with the given LSN, and as top transaction if that's specified);
650 * when this happens, is_new is set to true.
651 */
655{
656 ReorderBufferTXN *txn;
658 bool found;
659
661
662 /*
663 * Check the one-entry lookup cache first
664 */
666 rb->by_txn_last_xid == xid)
667 {
668 txn = rb->by_txn_last_txn;
669
671 {
672 /* found it, and it's valid */
675 return txn;
676 }
677
678 /*
679 * cached as non-existent, and asked not to create? Then nothing else
680 * to do.
681 */
682 if (!create)
684 /* otherwise fall through to create it */
685 }
686
687 /*
688 * If the cache wasn't hit or it yielded a "does-not-exist" and we want to
689 * create an entry.
690 */
691
692 /* search the lookup table */
695 &xid,
697 &found);
698 if (found)
699 txn = ent->txn;
700 else if (create)
701 {
702 /* initialize the new entry, if creation was requested */
705
707 ent->txn->xid = xid;
708 txn = ent->txn;
709 txn->first_lsn = lsn;
711
713 {
716 }
717 }
718 else
720
721 /* update cache */
722 rb->by_txn_last_xid = xid;
723 rb->by_txn_last_txn = txn;
724
726 *is_new = !found;
727
729 return txn;
730}
731
732/*
733 * Record the partial change for the streaming of in-progress transactions. We
734 * can stream only complete changes so if we have a partial change like toast
735 * table insert or speculative insert then we mark such a 'txn' so that it
736 * can't be streamed. We also ensure that if the changes in such a 'txn' can
737 * be streamed and are above logical_decoding_work_mem threshold then we stream
738 * them as soon as we have a complete change.
739 */
740static void
742 ReorderBufferChange *change,
744{
745 ReorderBufferTXN *toptxn;
746
747 /*
748 * The partial changes need to be processed only while streaming
749 * in-progress transactions.
750 */
752 return;
753
754 /* Get the top transaction. */
755 toptxn = rbtxn_get_toptxn(txn);
756
757 /*
758 * Indicate a partial change for toast inserts. The change will be
759 * considered as complete once we get the insert or update on the main
760 * table and we are sure that the pending toast chunks are not required
761 * anymore.
762 *
763 * If we allow streaming when there are pending toast chunks then such
764 * chunks won't be released till the insert (multi_insert) is complete and
765 * we expect the txn to have streamed all changes after streaming. This
766 * restriction is mainly to ensure the correctness of streamed
767 * transactions and it doesn't seem worth uplifting such a restriction
768 * just to allow this case because anyway we will stream the transaction
769 * once such an insert is complete.
770 */
777
778 /*
779 * Indicate a partial change for speculative inserts. The change will be
780 * considered as complete once we get the speculative confirm or abort
781 * token.
782 */
788
789 /*
790 * Stream the transaction if it is serialized before and the changes are
791 * now complete in the top-level transaction.
792 *
793 * The reason for doing the streaming of such a transaction as soon as we
794 * get the complete change for it is that previously it would have reached
795 * the memory threshold and wouldn't get streamed because of incomplete
796 * changes. Delaying such transactions would increase apply lag for them.
797 */
799 !(rbtxn_has_partial_change(toptxn)) &&
800 rbtxn_is_serialized(txn) &&
801 rbtxn_has_streamable_change(toptxn))
803}
804
805/*
806 * Queue a change into a transaction so it can be replayed upon commit or will be
807 * streamed when we reach logical_decoding_work_mem threshold.
808 */
809void
812{
813 ReorderBufferTXN *txn;
814
816
817 /*
818 * If we have detected that the transaction is aborted while streaming the
819 * previous changes or by checking its CLOG, there is no point in
820 * collecting further changes for it.
821 */
823 {
824 /*
825 * We don't need to update memory accounting for this change as we
826 * have not added it to the queue yet.
827 */
829 return;
830 }
831
832 /*
833 * The changes that are sent downstream are considered streamable. We
834 * remember such transactions so that only those will later be considered
835 * for streaming.
836 */
843 {
845
847 }
848
849 change->lsn = lsn;
850 change->txn = txn;
851
854 txn->nentries++;
855 txn->nentries_mem++;
856
857 /* update memory accounting information */
859 ReorderBufferChangeSize(change));
860
861 /* process partial change */
863
864 /* check the memory limits and evict something if needed */
866}
867
868/*
869 * A transactional message is queued to be processed upon commit and a
870 * non-transactional message gets processed immediately.
871 */
872void
875 bool transactional, const char *prefix,
877{
878 if (transactional)
879 {
880 MemoryContext oldcontext;
881 ReorderBufferChange *change;
882
884
885 /*
886 * We don't expect snapshots for transactional changes - we'll use the
887 * snapshot derived later during apply (unless the change gets
888 * skipped).
889 */
891
893
900
902
903 MemoryContextSwitchTo(oldcontext);
904 }
905 else
906 {
909
910 /* Non-transactional changes require a valid snapshot. */
911 Assert(snapshot_now);
912
915
916 /* setup snapshot to allow catalog access */
918 PG_TRY();
919 {
921
923 }
924 PG_CATCH();
925 {
927 PG_RE_THROW();
928 }
929 PG_END_TRY();
930 }
931}
932
933/*
934 * AssertTXNLsnOrder
935 * Verify LSN ordering of transaction lists in the reorderbuffer
936 *
937 * Other LSN-related invariants are checked too.
938 *
939 * No-op if assertions are not in use.
940 */
941static void
943{
944#ifdef USE_ASSERT_CHECKING
946 dlist_iter iter;
949
950 /*
951 * Skip the verification if we don't reach the LSN at which we start
952 * decoding the contents of transactions yet because until we reach the
953 * LSN, we could have transactions that don't have the association between
954 * the top-level transaction and subtransaction yet and consequently have
955 * the same LSN. We don't guarantee this association until we try to
956 * decode the actual contents of transaction. The ordering of the records
957 * prior to the start_decoding_at LSN should have been checked before the
958 * restart.
959 */
961 return;
962
964 {
966 iter.cur);
967
968 /* start LSN must be set */
970
971 /* If there is an end LSN, it must be higher than start LSN */
974
975 /* Current initial LSN must be strictly higher than previous */
978
979 /* known-as-subtxn txns must not be listed */
981
983 }
984
986 {
988 base_snapshot_node,
989 iter.cur);
990
991 /* base snapshot (and its LSN) must be set */
994
995 /* current LSN must be strictly higher than previous */
998
999 /* known-as-subtxn txns must not be listed */
1001
1003 }
1004#endif
1005}
1006
1007/*
1008 * AssertChangeLsnOrder
1009 *
1010 * Check ordering of changes in the (sub)transaction.
1011 */
1012static void
1014{
1015#ifdef USE_ASSERT_CHECKING
1016 dlist_iter iter;
1018
1020 {
1022
1024
1028
1031
1033
1035 }
1036#endif
1037}
1038
1039/*
1040 * ReorderBufferGetOldestTXN
1041 * Return oldest transaction in reorderbuffer
1042 */
1043ReorderBufferTXN *
1045{
1046 ReorderBufferTXN *txn;
1047
1049
1052
1054
1057 return txn;
1058}
1059
1060/*
1061 * ReorderBufferGetOldestXmin
1062 * Return oldest Xmin in reorderbuffer
1063 *
1064 * Returns oldest possibly running Xid from the point of view of snapshots
1065 * used in the transactions kept by reorderbuffer, or InvalidTransactionId if
1066 * there are none.
1067 *
1068 * Since snapshots are assigned monotonically, this equals the Xmin of the
1069 * base snapshot with minimal base_snapshot_lsn.
1070 */
1071TransactionId
1073{
1074 ReorderBufferTXN *txn;
1075
1077
1080
1082 &rb->txns_by_base_snapshot_lsn);
1084}
1085
1086void
1088{
1089 rb->current_restart_decoding_lsn = ptr;
1090}
1091
1092/*
1093 * ReorderBufferAssignChild
1094 *
1095 * Make note that we know that subxid is a subtransaction of xid, seen as of
1096 * the given lsn.
1097 */
1098void
1101{
1102 ReorderBufferTXN *txn;
1106
1109
1111 {
1113 {
1114 /* already associated, nothing to do */
1115 return;
1116 }
1117 else
1118 {
1119 /*
1120 * We already saw this transaction, but initially added it to the
1121 * list of top-level txns. Now that we know it's not top-level,
1122 * remove it from there.
1123 */
1125 }
1126 }
1127
1129 subtxn->toplevel_xid = xid;
1131
1132 /* set the reference to top-level transaction */
1133 subtxn->toptxn = txn;
1134
1135 /* add to subtransaction list */
1137 txn->nsubtxns++;
1138
1139 /* Possibly transfer the subtxn's snapshot to its top-level txn. */
1141
1142 /* Verify LSN-ordering invariant */
1144}
1145
1146/*
1147 * ReorderBufferTransferSnapToParent
1148 * Transfer base snapshot from subtxn to top-level txn, if needed
1149 *
1150 * This is done if the top-level txn doesn't have a base snapshot, or if the
1151 * subtxn's base snapshot has an earlier LSN than the top-level txn's base
1152 * snapshot's LSN. This can happen if there are no changes in the toplevel
1153 * txn but there are some in the subtxn, or the first change in subtxn has
1154 * earlier LSN than first change in the top-level txn and we learned about
1155 * their kinship only now.
1156 *
1157 * The subtransaction's snapshot is cleared regardless of the transfer
1158 * happening, since it's not needed anymore in either case.
1159 *
1160 * We do this as soon as we become aware of their kinship, to avoid queueing
1161 * extra snapshots to txns known-as-subtxns -- only top-level txns will
1162 * receive further snapshots.
1163 */
1164static void
1167{
1169
1171 {
1174 {
1175 /*
1176 * If the toplevel transaction already has a base snapshot but
1177 * it's newer than the subxact's, purge it.
1178 */
1180 {
1183 }
1184
1185 /*
1186 * The snapshot is now the top transaction's; transfer it, and
1187 * adjust the list position of the top transaction in the list by
1188 * moving it to where the subtransaction is.
1189 */
1193 &txn->base_snapshot_node);
1194
1195 /*
1196 * The subtransaction doesn't have a snapshot anymore (so it
1197 * mustn't be in the list.)
1198 */
1202 }
1203 else
1204 {
1205 /* Base snap of toplevel is fine, so subxact's is not needed */
1210 }
1211 }
1212}
1213
1214/*
1215 * Associate a subtransaction with its toplevel transaction at commit
1216 * time. There may be no further changes added after this.
1217 */
1218void
1221 XLogRecPtr end_lsn)
1222{
1224
1227
1228 /*
1229 * No need to do anything if that subtxn didn't contain any changes
1230 */
1232 return;
1233
1234 subtxn->final_lsn = commit_lsn;
1235 subtxn->end_lsn = end_lsn;
1236
1237 /*
1238 * Assign this subxact as a child of the toplevel xact (no-op if already
1239 * done.)
1240 */
1242}
1243
1244
1245/*
1246 * Support for efficiently iterating over a transaction's and its
1247 * subtransactions' changes.
1248 *
1249 * We do by doing a k-way merge between transactions/subtransactions. For that
1250 * we model the current heads of the different transactions as a binary heap
1251 * so we easily know which (sub-)transaction has the change with the smallest
1252 * lsn next.
1253 *
1254 * We assume the changes in individual transactions are already sorted by LSN.
1255 */
1256
1257/*
1258 * Binary heap comparison function.
1259 */
1260static int
1262{
1266
1268 return 1;
1270 return 0;
1271 return -1;
1272}
1273
1274/*
1275 * Allocate & initialize an iterator which iterates in lsn order over a
1276 * transaction and all its subtransactions.
1277 *
1278 * Note: The iterator state is returned through iter_state parameter rather
1279 * than the function's return value. This is because the state gets cleaned up
1280 * in a PG_CATCH block in the caller, so we want to make sure the caller gets
1281 * back the state even if this function throws an exception.
1282 */
1283static void
1286{
1287 Size nr_txns = 0;
1290 Size off;
1291
1293
1294 /* Check ordering of changes in the toplevel transaction. */
1295 AssertChangeLsnOrder(txn);
1296
1297 /*
1298 * Calculate the size of our heap: one element for every transaction that
1299 * contains changes. (Besides the transactions already in the reorder
1300 * buffer, we count the one we were directly passed.)
1301 */
1303 nr_txns++;
1304
1306 {
1308
1310
1311 /* Check ordering of changes in this subtransaction. */
1313
1315 nr_txns++;
1316 }
1317
1318 /* allocate iteration state */
1323
1324 state->nr_txns = nr_txns;
1326
1328 {
1329 state->entries[off].file.vfd = -1;
1330 state->entries[off].segno = 0;
1331 }
1332
1333 /* allocate heap */
1335 ReorderBufferIterCompare,
1336 state);
1337
1338 /* Now that the state fields are initialized, it is safe to return it. */
1340
1341 /*
1342 * Now insert items into the binary heap, in an unordered fashion. (We
1343 * will run a heap assembly step at the end; this is more efficient.)
1344 */
1345
1346 off = 0;
1347
1348 /* add toplevel transaction if it contains changes */
1350 {
1352
1354 {
1355 /* serialize remaining changes */
1358 &state->entries[off].segno);
1359 }
1360
1362 &txn->changes);
1363
1366 state->entries[off].txn = txn;
1367
1369 }
1370
1371 /* add subtransactions if they contain changes */
1373 {
1375
1377
1379 {
1381
1383 {
1384 /* serialize remaining changes */
1387 &state->entries[off].file,
1388 &state->entries[off].segno);
1389 }
1391 &cur_txn->changes);
1392
1396
1398 }
1399 }
1400
1401 /* assemble a valid binary heap */
1403}
1404
1405/*
1406 * Return the next change when iterating over a transaction and its
1407 * subtransactions.
1408 *
1409 * Returns NULL when no further changes exist.
1410 */
1413{
1414 ReorderBufferChange *change;
1415 ReorderBufferIterTXNEntry *entry;
1416 int32 off;
1417
1418 /* nothing there anymore */
1421
1423 entry = &state->entries[off];
1424
1425 /* free memory we might have "leaked" in the previous *Next call */
1427 {
1432 }
1433
1434 change = entry->change;
1435
1436 /*
1437 * update heap with information about which transaction has the next
1438 * relevant change in LSN order
1439 */
1440
1441 /* there are in-memory changes */
1443 {
1447
1448 /* txn stays the same */
1451
1453 return change;
1454 }
1455
1456 /* try to load changes from disk */
1458 {
1459 /*
1460 * Ugly: restoring changes will reuse *Change records, thus delete the
1461 * current one from the per-tx list and only free in the next call.
1462 */
1465
1466 /*
1467 * Update the total bytes processed by the txn for which we are
1468 * releasing the current set of changes and restoring the new set of
1469 * changes.
1470 */
1473 &state->entries[off].segno))
1474 {
1475 /* successfully restored changes from disk */
1479
1483
1485 /* txn stays the same */
1489
1490 return change;
1491 }
1492 }
1493
1494 /* ok, no changes there anymore, remove */
1496
1497 return change;
1498}
1499
1500/*
1501 * Deallocate the iterator
1502 */
1503static void
1506{
1507 Size off;
1508
1510 {
1513 }
1514
1515 /* free memory we might have "leaked" in the last *Next call */
1517 {
1518 ReorderBufferChange *change;
1519
1524 }
1525
1528}
1529
1530/*
1531 * Cleanup the contents of a transaction, usually after the transaction
1532 * committed or aborted.
1533 */
1534static void
1536{
1537 bool found;
1538 dlist_mutable_iter iter;
1540
1541 /* cleanup subtransactions & their changes */
1543 {
1545
1547
1548 /*
1549 * Subtransactions are always associated to the toplevel TXN, even if
1550 * they originally were happening inside another subtxn, so we won't
1551 * ever recurse more than one level deep here.
1552 */
1555
1557 }
1558
1559 /* cleanup changes in the txn */
1561 {
1562 ReorderBufferChange *change;
1563
1565
1566 /* Check we're not mixing changes from different transactions. */
1568
1569 /*
1570 * Instead of updating the memory counter for individual changes, we
1571 * sum up the size of memory to free so we can update the memory
1572 * counter all together below. This saves costs of maintaining the
1573 * max-heap.
1574 */
1576
1578 }
1579
1580 /* Update the memory counter */
1582
1583 /*
1584 * Cleanup the tuplecids we stored for decoding catalog snapshot access.
1585 * They are always stored in the toplevel transaction.
1586 */
1588 {
1589 ReorderBufferChange *change;
1590
1592
1593 /* Check we're not mixing changes from different transactions. */
1596
1598 }
1599
1600 /*
1601 * Cleanup the base snapshot, if set.
1602 */
1604 {
1607 }
1608
1609 /*
1610 * Cleanup the snapshot for the last streamed run.
1611 */
1613 {
1616 }
1617
1618 /*
1619 * Remove TXN from its containing lists.
1620 *
1621 * Note: if txn is known as subxact, we are deleting the TXN from its
1622 * parent's list of known subxacts; this leaves the parent's nsubxacts
1623 * count too high, but we don't care. Otherwise, we are deleting the TXN
1624 * from the LSN-ordered list of toplevel TXNs. We remove the TXN from the
1625 * list of catalog modifying transactions as well.
1626 */
1630
1631 /* now remove reference from buffer */
1633 Assert(found);
1634
1635 /* remove entries spilled to disk */
1638
1639 /* deallocate */
1641}
1642
1643/*
1644 * Discard changes from a transaction (and subtransactions), either after
1645 * streaming, decoding them at PREPARE, or detecting the transaction abort.
1646 * Keep the remaining info - transactions, tuplecids, invalidations and
1647 * snapshots.
1648 *
1649 * We additionally remove tuplecids after decoding the transaction at prepare
1650 * time as we only need to perform invalidation at rollback or commit prepared.
1651 *
1652 * 'txn_prepared' indicates that we have decoded the transaction at prepare
1653 * time.
1654 */
1655static void
1657{
1658 dlist_mutable_iter iter;
1660
1661 /* cleanup subtransactions & their changes */
1663 {
1665
1667
1668 /*
1669 * Subtransactions are always associated to the toplevel TXN, even if
1670 * they originally were happening inside another subtxn, so we won't
1671 * ever recurse more than one level deep here.
1672 */
1675
1678 }
1679
1680 /* cleanup changes in the txn */
1682 {
1683 ReorderBufferChange *change;
1684
1686
1687 /* Check we're not mixing changes from different transactions. */
1689
1690 /* remove the change from its containing list */
1692
1693 /*
1694 * Instead of updating the memory counter for individual changes, we
1695 * sum up the size of memory to free so we can update the memory
1696 * counter all together below. This saves costs of maintaining the
1697 * max-heap.
1698 */
1700
1702 }
1703
1704 /* Update the memory counter */
1706
1708 {
1709 /*
1710 * If this is a prepared txn, cleanup the tuplecids we stored for
1711 * decoding catalog snapshot access. They are always stored in the
1712 * toplevel transaction.
1713 */
1715 {
1716 ReorderBufferChange *change;
1717
1719
1720 /* Check we're not mixing changes from different transactions. */
1723
1724 /* Remove the change from its containing list. */
1726
1728 }
1729 }
1730
1731 /*
1732 * Destroy the (relfilelocator, ctid) hashtable, so that we don't leak any
1733 * memory. We could also keep the hash table and update it with new ctid
1734 * values, but this seems simpler and good enough for now.
1735 */
1737 {
1740 }
1741
1742 /* If this txn is serialized then clean the disk space. */
1744 {
1747
1748 /*
1749 * We set this flag to indicate if the transaction is ever serialized.
1750 * We need this to accurately update the stats as otherwise the same
1751 * transaction can be counted as serialized multiple times.
1752 */
1754 }
1755
1756 /* also reset the number of entries in the transaction */
1757 txn->nentries_mem = 0;
1758 txn->nentries = 0;
1759}
1760
1761/*
1762 * Check the transaction status by CLOG lookup and discard all changes if
1763 * the transaction is aborted. The transaction status is cached in
1764 * txn->txn_flags so we can skip future changes and avoid CLOG lookups on the
1765 * next call.
1766 *
1767 * Return true if the transaction is aborted, otherwise return false.
1768 *
1769 * When the 'debug_logical_replication_streaming' is set to "immediate", we
1770 * don't check the transaction status, meaning the caller will always process
1771 * this transaction.
1772 */
1773static bool
1775{
1776 /* Quick return for regression tests */
1778 return false;
1779
1780 /*
1781 * Quick return if the transaction status is already known.
1782 */
1783
1785 return false;
1787 {
1788 /* Already-aborted transactions should not have any changes */
1790
1791 return true;
1792 }
1793
1794 /* Otherwise, check the transaction status using CLOG lookup */
1795
1797 return false;
1798
1800 {
1801 /*
1802 * Remember the transaction is committed so that we can skip CLOG
1803 * check next time, avoiding the pressure on CLOG lookup.
1804 */
1807 return false;
1808 }
1809
1810 /*
1811 * The transaction aborted. We discard both the changes collected so far
1812 * and the toast reconstruction data. The full cleanup will happen as part
1813 * of decoding ABORT record of this transaction.
1814 */
1817
1818 /* All changes should be discarded */
1820
1821 /*
1822 * Mark the transaction as aborted so we can ignore future changes of this
1823 * transaction.
1824 */
1827
1828 return true;
1829}
1830
1831/*
1832 * Build a hash with a (relfilelocator, ctid) -> (cmin, cmax) mapping for use by
1833 * HeapTupleSatisfiesHistoricMVCC.
1834 */
1835static void
1837{
1838 dlist_iter iter;
1840
1842 return;
1843
1847
1848 /*
1849 * create the hash with the exact number of to-be-stored tuplecids from
1850 * the start
1851 */
1852 txn->tuplecid_hash =
1855
1857 {
1860 bool found;
1861 ReorderBufferChange *change;
1862
1864
1866
1867 /* be careful about padding */
1869
1871
1873 &key.tid);
1874
1877 if (!found)
1878 {
1882 }
1883 else
1884 {
1885 /*
1886 * Maybe we already saw this tuple before in this transaction, but
1887 * if so it must have the same cmin.
1888 */
1890
1891 /*
1892 * cmax may be initially invalid, but once set it can only grow,
1893 * and never become invalid again.
1894 */
1899 }
1900 }
1901}
1902
1903/*
1904 * Copy a provided snapshot so we can modify it privately. This is needed so
1905 * that catalog modifying transactions can look into intermediate catalog
1906 * states.
1907 */
1911{
1913 dlist_iter iter;
1915 Size size;
1916
1920
1923
1926 snap->regd_count = 0;
1928
1930
1931 /*
1932 * snap->subxip contains all txids that belong to our transaction which we
1933 * need to check via cmin/cmax. That's why we store the toplevel
1934 * transaction in there as well.
1935 */
1938
1939 /*
1940 * txn->nsubtxns isn't decreased when subtransactions abort, so count
1941 * manually. Since it's an upper boundary it is safe to use it for the
1942 * allocation above.
1943 */
1944 snap->subxcnt = 1;
1945
1947 {
1949
1952 snap->subxcnt++;
1953 }
1954
1955 /* sort so we can bsearch() later */
1957
1958 /* store the specified current CommandId */
1960
1962}
1963
1964/*
1965 * Free a previously ReorderBufferCopySnap'ed snapshot
1966 */
1967static void
1969{
1972 else
1974}
1975
1976/*
1977 * If the transaction was (partially) streamed, we need to prepare or commit
1978 * it in a 'streamed' way. That is, we first stream the remaining part of the
1979 * transaction, and then invoke stream_prepare or stream_commit message as per
1980 * the case.
1981 */
1982static void
1984{
1985 /* we should only call this for previously streamed transactions */
1987
1989
1991 {
1992 /*
1993 * Note, we send stream prepare even if a concurrent abort is
1994 * detected. See DecodePrepare for more information.
1995 */
1999
2000 /*
2001 * This is a PREPARED transaction, part of a two-phase commit. The
2002 * full cleanup will happen as part of the COMMIT PREPAREDs, so now
2003 * just truncate txn by removing changes and tuplecids.
2004 */
2006 /* Reset the CheckXidAlive */
2008 }
2009 else
2010 {
2013 }
2014}
2015
2016/*
2017 * Set xid to detect concurrent aborts.
2018 *
2019 * While streaming an in-progress transaction or decoding a prepared
2020 * transaction there is a possibility that the (sub)transaction might get
2021 * aborted concurrently. In such case if the (sub)transaction has catalog
2022 * update then we might decode the tuple using wrong catalog version. For
2023 * example, suppose there is one catalog tuple with (xmin: 500, xmax: 0). Now,
2024 * the transaction 501 updates the catalog tuple and after that we will have
2025 * two tuples (xmin: 500, xmax: 501) and (xmin: 501, xmax: 0). Now, if 501 is
2026 * aborted and some other transaction say 502 updates the same catalog tuple
2027 * then the first tuple will be changed to (xmin: 500, xmax: 502). So, the
2028 * problem is that when we try to decode the tuple inserted/updated in 501
2029 * after the catalog update, we will see the catalog tuple with (xmin: 500,
2030 * xmax: 502) as visible because it will consider that the tuple is deleted by
2031 * xid 502 which is not visible to our snapshot. And when we will try to
2032 * decode with that catalog tuple, it can lead to a wrong result or a crash.
2033 * So, it is necessary to detect concurrent aborts to allow streaming of
2034 * in-progress transactions or decoding of prepared transactions.
2035 *
2036 * For detecting the concurrent abort we set CheckXidAlive to the current
2037 * (sub)transaction's xid for which this change belongs to. And, during
2038 * catalog scan we can check the status of the xid and if it is aborted we will
2039 * report a specific error so that we can stop streaming current transaction
2040 * and discard the already streamed changes on such an error. We might have
2041 * already streamed some of the changes for the aborted (sub)transaction, but
2042 * that is fine because when we decode the abort we will stream abort message
2043 * to truncate the changes in the subscriber. Similarly, for prepared
2044 * transactions, we stop decoding if concurrent abort is detected and then
2045 * rollback the changes when rollback prepared is encountered. See
2046 * DecodePrepare.
2047 */
2048static inline void
2050{
2051 /*
2052 * If the input transaction id is already set as a CheckXidAlive then
2053 * nothing to do.
2054 */
2056 return;
2057
2058 /*
2059 * setup CheckXidAlive if it's not committed yet. We don't check if the
2060 * xid is aborted. That will happen during catalog access.
2061 */
2063 CheckXidAlive = xid;
2064 else
2066}
2067
2068/*
2069 * Helper function for ReorderBufferProcessTXN for applying change.
2070 */
2071static inline void
2074 bool streaming)
2075{
2076 if (streaming)
2078 else
2080}
2081
2082/*
2083 * Helper function for ReorderBufferProcessTXN for applying the truncate.
2084 */
2085static inline void
2089{
2090 if (streaming)
2092 else
2094}
2095
2096/*
2097 * Helper function for ReorderBufferProcessTXN for applying the message.
2098 */
2099static inline void
2102{
2103 if (streaming)
2108 else
2113}
2114
2115/*
2116 * Function to store the command id and snapshot at the end of the current
2117 * stream so that we can reuse the same while sending the next stream.
2118 */
2119static inline void
2122{
2123 txn->command_id = command_id;
2124
2125 /* Avoid copying if it's already copied. */
2127 txn->snapshot_now = snapshot_now;
2128 else
2130 txn, command_id);
2131}
2132
2133/*
2134 * Mark the given transaction as streamed if it's a top-level transaction
2135 * or has changes.
2136 */
2137static void
2139{
2140 /*
2141 * The top-level transaction, is marked as streamed always, even if it
2142 * does not contain any changes (that is, when all the changes are in
2143 * subtransactions).
2144 *
2145 * For subtransactions, we only mark them as streamed when there are
2146 * changes in them.
2147 *
2148 * We do it this way because of aborts - we don't want to send aborts for
2149 * XIDs the downstream is not aware of. And of course, it always knows
2150 * about the top-level xact (we send the XID in all messages), but we
2151 * never stream XIDs of empty subxacts.
2152 */
2155}
2156
2157/*
2158 * Helper function for ReorderBufferProcessTXN to handle the concurrent
2159 * abort of the streaming transaction. This resets the TXN such that it
2160 * can be used to stream the remaining data of transaction being processed.
2161 * This can happen when the subtransaction is aborted and we still want to
2162 * continue processing the main or other subtransactions data.
2163 */
2164static void
2166 Snapshot snapshot_now,
2167 CommandId command_id,
2168 XLogRecPtr last_lsn)
2169{
2170 /* Discard the changes that we just streamed */
2172
2173 /* Free all resources allocated for toast reconstruction */
2175
2176 /*
2177 * For the streaming case, stop the stream and remember the command ID and
2178 * snapshot for the streaming run.
2179 */
2181 {
2184 }
2185
2186 /* All changes must be deallocated */
2188}
2189
2190/*
2191 * Helper function for ReorderBufferReplay and ReorderBufferStreamTXN.
2192 *
2193 * Send data of a transaction (and its subtransactions) to the
2194 * output plugin. We iterate over the top and subtransactions (using a k-way
2195 * merge) and replay the changes in lsn order.
2196 *
2197 * If streaming is true then data will be sent using stream API.
2198 *
2199 * Note: "volatile" markers on some parameters are to avoid trouble with
2200 * PG_TRY inside the function.
2201 */
2202static void
2204 XLogRecPtr commit_lsn,
2207 bool streaming)
2208{
2217
2218 /* build data to be able to lookup the CommandIds of catalog tuples */
2220
2221 /* setup the initial snapshot */
2223
2224 /*
2225 * Decoding needs access to syscaches et al., which in turn use
2226 * heavyweight locks and such. Thus we need to have enough state around to
2227 * keep track of those. The easiest way is to simply use a transaction
2228 * internally. That also allows us to easily enforce that nothing writes
2229 * to the database by checking for xid assignments.
2230 *
2231 * When we're called via the SQL SRF there's already a transaction
2232 * started, so start an explicit subtransaction there.
2233 */
2235
2236 PG_TRY();
2237 {
2238 ReorderBufferChange *change;
2240 * changes */
2241
2244 else
2245 StartTransactionCommand();
2246
2247 /*
2248 * We only need to send begin/begin-prepare for non-streamed
2249 * transactions.
2250 */
2251 if (!streaming)
2252 {
2255 else
2257 }
2258
2261 {
2263 Oid reloid;
2264
2265 CHECK_FOR_INTERRUPTS();
2266
2267 /*
2268 * We can't call start stream callback before processing first
2269 * change.
2270 */
2272 {
2273 if (streaming)
2274 {
2278 }
2279 }
2280
2281 /*
2282 * Enforce correct ordering of changes, merged from multiple
2283 * subtransactions. The changes may have the same LSN due to
2284 * MULTI_INSERT xlog records.
2285 */
2287
2289
2290 /*
2291 * Set the current xid to detect concurrent aborts. This is
2292 * required for the cases when we decode the changes before the
2293 * COMMIT record is processed.
2294 */
2296 {
2299 }
2300
2302 {
2304
2305 /*
2306 * Confirmation for speculative insertion arrived. Simply
2307 * use as a normal record. It'll be cleaned up at the end
2308 * of INSERT processing.
2309 */
2313 change = specinsert;
2315
2316 /* intentionally fall through */
2317 pg_fallthrough;
2321 Assert(snapshot_now);
2322
2325
2326 /*
2327 * Mapped catalog tuple without data, emitted while
2328 * catalog table was in the process of being rewritten. We
2329 * can fail to look up the relfilenumber, because the
2330 * relmapper has no "historic" view, in contrast to the
2331 * normal catalog during decoding. Thus repeated rewrites
2332 * can cause a lookup failure. That's OK because we do not
2333 * decode catalog changes anyway. Normally such tuples
2334 * would be skipped over below, but we can't identify
2335 * whether the table should be logically logged without
2336 * mapping the relfilenumber to the oid.
2337 */
2345 MAIN_FORKNUM).str);
2346
2347 relation = RelationIdGetRelation(reloid);
2348
2351 reloid,
2353 MAIN_FORKNUM).str);
2354
2357
2358 /*
2359 * Ignore temporary heaps created during DDL unless the
2360 * plugin has asked for them.
2361 */
2364
2365 /*
2366 * For now ignore sequence changes entirely. Most of the
2367 * time they don't log changes using records we
2368 * understand, so it doesn't make sense to handle the few
2369 * cases we do.
2370 */
2373
2374 /* user-triggered change */
2376 {
2379 streaming);
2380
2381 /*
2382 * Only clear reassembled toast chunks if we're sure
2383 * they're not required anymore. The creator of the
2384 * tuple tells us.
2385 */
2388 }
2389 /* we're not interested in toast deletions */
2391 {
2392 /*
2393 * Need to reassemble the full toasted Datum in
2394 * memory, to ensure the chunks don't get reused till
2395 * we're done remove it from the list of this
2396 * transaction's changes. Otherwise it will get
2397 * freed/reused while restoring spooled data from
2398 * disk.
2399 */
2401
2404 change);
2405 }
2406
2407 change_done:
2408
2409 /*
2410 * If speculative insertion was confirmed, the record
2411 * isn't needed anymore.
2412 */
2414 {
2417 }
2418
2420 {
2421 RelationClose(relation);
2422 relation = NULL;
2423 }
2424 break;
2425
2427
2428 /*
2429 * Speculative insertions are dealt with by delaying the
2430 * processing of the insert until the confirmation record
2431 * arrives. For that we simply unlink the record from the
2432 * chain, so it does not get freed/reused while restoring
2433 * spooled data from disk.
2434 *
2435 * This is safe in the face of concurrent catalog changes
2436 * because the relevant relation can't be changed between
2437 * speculative insertion and confirmation due to
2438 * CheckTableNotInUse() and locking.
2439 */
2440
2441 /* Previous speculative insertion must be aborted */
2443
2444 /* and memorize the pending insertion */
2446 specinsert = change;
2447 break;
2448
2450
2451 /*
2452 * Abort for speculative insertion arrived. So cleanup the
2453 * specinsert tuple and toast hash.
2454 *
2455 * Note that we get the spec abort change for each toast
2456 * entry but we need to perform the cleanup only the first
2457 * time we get it for the main table.
2458 */
2460 {
2461 /*
2462 * We must clean the toast hash before processing a
2463 * completely new tuple to avoid confusion about the
2464 * previous tuple's toast chunks.
2465 */
2468
2469 /* We don't need this record anymore. */
2472 }
2473 break;
2474
2476 {
2480 Relation *relations;
2481
2484 {
2486 Relation rel;
2487
2488 rel = RelationIdGetRelation(relid);
2489
2492
2494 continue;
2495
2496 relations[nrelations++] = rel;
2497 }
2498
2499 /* Apply the truncate. */
2501 relations, change,
2502 streaming);
2503
2506
2507 break;
2508 }
2509
2512 break;
2513
2515 /* Execute the invalidation messages locally */
2518 break;
2519
2521 /* get rid of the old */
2523
2525 {
2527 snapshot_now =
2529 txn, command_id);
2530 }
2531
2532 /*
2533 * Restored from disk, need to be careful not to double
2534 * free. We could introduce refcounting for that, but for
2535 * now this seems infrequent enough not to care.
2536 */
2538 {
2539 snapshot_now =
2541 txn, command_id);
2542 }
2543 else
2544 {
2546 }
2547
2548 /* and continue with the new one */
2550 break;
2551
2554
2556 {
2558
2560 {
2561 /* we don't use the global one anymore */
2563 txn, command_id);
2564 }
2565
2566 snapshot_now->curcid = command_id;
2567
2570 }
2571
2572 break;
2573
2576 break;
2577 }
2578
2579 /*
2580 * It is possible that the data is not sent to downstream for a
2581 * long time either because the output plugin filtered it or there
2582 * is a DDL that generates a lot of data that is not processed by
2583 * the plugin. So, in such cases, the downstream can timeout. To
2584 * avoid that we try to send a keepalive message if required.
2585 * Trying to send a keepalive message after every change has some
2586 * overhead, but testing showed there is no noticeable overhead if
2587 * we do it after every ~100 changes.
2588 */
2589#define CHANGES_THRESHOLD 100
2590
2592 {
2594 changes_count = 0;
2595 }
2596 }
2597
2598 /* speculative insertion record must be freed by now */
2600
2601 /* clean up the iterator */
2604
2605 /*
2606 * Update total transaction count and total bytes processed by the
2607 * transaction and its subtransactions. Ensure to not count the
2608 * streamed transaction multiple times.
2609 *
2610 * Note that the statistics computation has to be done after
2611 * ReorderBufferIterTXNFinish as it releases the serialized change
2612 * which we have already accounted in ReorderBufferIterTXNNext.
2613 */
2615 rb->totalTxns++;
2616
2618
2619 /*
2620 * Done with current changes, send the last message for this set of
2621 * changes depending upon streaming mode.
2622 */
2623 if (streaming)
2624 {
2626 {
2629 }
2630 }
2631 else
2632 {
2633 /*
2634 * Call either PREPARE (for two-phase transactions) or COMMIT (for
2635 * regular ones).
2636 */
2638 {
2642 }
2643 else
2645 }
2646
2647 /* this is just a sanity check against bad output plugin behaviour */
2650 GetCurrentTransactionId());
2651
2652 /*
2653 * Remember the command ID and snapshot for the next set of changes in
2654 * streaming mode.
2655 */
2656 if (streaming)
2660
2661 /* cleanup */
2663
2664 /*
2665 * Aborting the current (sub-)transaction as a whole has the right
2666 * semantics. We want all locks acquired in here to be released, not
2667 * reassigned to the parent and we do not want any database access
2668 * have persistent effects.
2669 */
2670 AbortCurrentTransaction();
2671
2672 /* make sure there's no cache pollution */
2674 {
2676 InvalidateSystemCaches();
2677 }
2678 else
2679 {
2682 txn->invalidations_distributed);
2683 }
2684
2686 {
2689 CurrentResourceOwner = cowner;
2690 }
2691
2692 /*
2693 * We are here due to one of the four reasons: 1. Decoding an
2694 * in-progress txn. 2. Decoding a prepared txn. 3. Decoding of a
2695 * prepared txn that was (partially) streamed. 4. Decoding a committed
2696 * txn.
2697 *
2698 * For 1, we allow truncation of txn data by removing the changes
2699 * already streamed but still keeping other things like invalidations,
2700 * snapshot, and tuplecids. For 2 and 3, we indicate
2701 * ReorderBufferTruncateTXN to do more elaborate truncation of txn
2702 * data as the entire transaction has been decoded except for commit.
2703 * For 4, as the entire txn has been decoded, we can fully clean up
2704 * the TXN reorder buffer.
2705 */
2707 {
2708 if (streaming)
2710
2712 /* Reset the CheckXidAlive */
2714 }
2715 else
2717 }
2718 PG_CATCH();
2719 {
2722
2723 /* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */
2726
2728
2729 /*
2730 * Force cache invalidation to happen outside of a valid transaction
2731 * to prevent catalog access as we just caught an error.
2732 */
2733 AbortCurrentTransaction();
2734
2735 /* make sure there's no cache pollution */
2737 {
2739 InvalidateSystemCaches();
2740 }
2741 else
2742 {
2745 txn->invalidations_distributed);
2746 }
2747
2749 {
2752 CurrentResourceOwner = cowner;
2753 }
2754
2755 /* Free the specinsert change before freeing the ReorderBufferTXN */
2757 {
2760 }
2761
2762 /*
2763 * The error code ERRCODE_TRANSACTION_ROLLBACK indicates a concurrent
2764 * abort of the (sub)transaction we are streaming or preparing. We
2765 * need to do the cleanup and return gracefully on this error, see
2766 * SetupCheckXidLive.
2767 *
2768 * This error code can be thrown by one of the callbacks we call
2769 * during decoding so we need to ensure that we return gracefully only
2770 * when we are sending the data in streaming mode and the streaming is
2771 * not finished yet or when we are sending the data out on a PREPARE
2772 * during a two-phase commit.
2773 */
2776 {
2777 /* curtxn must be set for streaming or prepared transactions */
2779
2780 /* Cleanup the temporary error state. */
2781 FlushErrorState();
2784
2785 /* Remember the transaction is aborted. */
2788
2789 /* Mark the transaction is streamed if appropriate */
2792
2793 /* Reset the TXN so that it is allowed to stream remaining data. */
2795 command_id, prev_lsn);
2796 }
2797 else
2798 {
2801 PG_RE_THROW();
2802 }
2803 }
2804 PG_END_TRY();
2805}
2806
2807/*
2808 * Perform the replay of a transaction and its non-aborted subtransactions.
2809 *
2810 * Subtransactions previously have to be processed by
2811 * ReorderBufferCommitChild(), even if previously assigned to the toplevel
2812 * transaction with ReorderBufferAssignChild.
2813 *
2814 * This interface is called once a prepare or toplevel commit is read for both
2815 * streamed as well as non-streamed transactions.
2816 */
2817static void
2821 TimestampTz commit_time,
2823{
2824 Snapshot snapshot_now;
2826
2827 txn->final_lsn = commit_lsn;
2828 txn->end_lsn = end_lsn;
2829 txn->commit_time = commit_time;
2830 txn->origin_id = origin_id;
2831 txn->origin_lsn = origin_lsn;
2832
2833 /*
2834 * If the transaction was (partially) streamed, we need to commit it in a
2835 * 'streamed' way. That is, we first stream the remaining part of the
2836 * transaction, and then invoke stream_commit message.
2837 *
2838 * Called after everything (origin ID, LSN, ...) is stored in the
2839 * transaction to avoid passing that information directly.
2840 */
2842 {
2844 return;
2845 }
2846
2847 /*
2848 * If this transaction has no snapshot, it didn't make any changes to the
2849 * database, so there's nothing to decode. Note that
2850 * ReorderBufferCommitChild will have transferred any snapshots from
2851 * subtransactions if there were any.
2852 */
2854 {
2856
2857 /*
2858 * Removing this txn before a commit might result in the computation
2859 * of an incorrect restart_lsn. See SnapBuildProcessRunningXacts.
2860 */
2863 return;
2864 }
2865
2866 snapshot_now = txn->base_snapshot;
2867
2868 /* Process and send the changes to output plugin. */
2870 command_id, false);
2871}
2872
2873/*
2874 * Commit a transaction.
2875 *
2876 * See comments for ReorderBufferReplay().
2877 */
2878void
2881 TimestampTz commit_time,
2883{
2884 ReorderBufferTXN *txn;
2885
2887 false);
2888
2889 /* unknown transaction, nothing to replay */
2891 return;
2892
2894 origin_id, origin_lsn);
2895}
2896
2897/*
2898 * Record the prepare information for a transaction. Also, mark the transaction
2899 * as a prepared transaction.
2900 */
2901bool
2904 TimestampTz prepare_time,
2906{
2907 ReorderBufferTXN *txn;
2908
2910
2911 /* unknown transaction, nothing to do */
2913 return false;
2914
2915 /*
2916 * Remember the prepare information to be later used by commit prepared in
2917 * case we skip doing prepare.
2918 */
2919 txn->final_lsn = prepare_lsn;
2920 txn->end_lsn = end_lsn;
2921 txn->prepare_time = prepare_time;
2922 txn->origin_id = origin_id;
2923 txn->origin_lsn = origin_lsn;
2924
2925 /* Mark this transaction as a prepared transaction */
2928
2929 return true;
2930}
2931
2932/* Remember that we have skipped prepare */
2933void
2935{
2936 ReorderBufferTXN *txn;
2937
2939
2940 /* unknown transaction, nothing to do */
2942 return;
2943
2944 /* txn must have been marked as a prepared transaction */
2947}
2948
2949/*
2950 * Prepare a two-phase transaction.
2951 *
2952 * See comments for ReorderBufferReplay().
2953 */
2954void
2956 char *gid)
2957{
2958 ReorderBufferTXN *txn;
2959
2961 false);
2962
2963 /* unknown transaction, nothing to replay */
2965 return;
2966
2967 /*
2968 * txn must have been marked as a prepared transaction and must have
2969 * neither been skipped nor sent a prepare. Also, the prepare info must
2970 * have been updated in it by now.
2971 */
2974
2976
2979
2980 /*
2981 * Send a prepare if not already done so. This might occur if we have
2982 * detected a concurrent abort while replaying the non-streaming
2983 * transaction.
2984 */
2986 {
2989 }
2990}
2991
2992/*
2993 * This is used to handle COMMIT/ROLLBACK PREPARED.
2994 */
2995void
2998 XLogRecPtr two_phase_at,
3001{
3002 ReorderBufferTXN *txn;
3003 XLogRecPtr prepare_end_lsn;
3004 TimestampTz prepare_time;
3005
3007
3008 /* unknown transaction, nothing to do */
3010 return;
3011
3012 /*
3013 * By this time the txn has the prepare record information, remember it to
3014 * be later used for rollback.
3015 */
3016 prepare_end_lsn = txn->end_lsn;
3017 prepare_time = txn->prepare_time;
3018
3019 /* add the gid in the txn */
3021
3022 /*
3023 * It is possible that this transaction is not decoded at prepare time
3024 * either because by that time we didn't have a consistent snapshot, or
3025 * two_phase was not enabled, or it was decoded earlier but we have
3026 * restarted. We only need to send the prepare if it was not decoded
3027 * earlier. We don't need to decode the xact for aborts if it is not done
3028 * already.
3029 */
3031 {
3032 /*
3033 * txn must have been marked as a prepared transaction and skipped but
3034 * not sent a prepare. Also, the prepare info must have been updated
3035 * in txn even if we skip prepare.
3036 */
3040
3041 /*
3042 * By this time the txn has the prepare record information and it is
3043 * important to use that so that downstream gets the accurate
3044 * information. If instead, we have passed commit information here
3045 * then downstream can behave as it has already replayed commit
3046 * prepared after the restart.
3047 */
3050 }
3051
3052 txn->final_lsn = commit_lsn;
3053 txn->end_lsn = end_lsn;
3054 txn->commit_time = commit_time;
3055 txn->origin_id = origin_id;
3056 txn->origin_lsn = origin_lsn;
3057
3060 else
3062
3063 /* cleanup: make sure there's no cache pollution */
3065 txn->invalidations);
3067}
3068
3069/*
3070 * Abort a transaction that possibly has previous changes. Needs to be first
3071 * called for subtransactions and then for the toplevel xid.
3072 *
3073 * NB: Transactions handled here have to have actively aborted (i.e. have
3074 * produced an abort record). Implicitly aborted transactions are handled via
3075 * ReorderBufferAbortOld(); transactions we're just not interested in, but
3076 * which have committed are handled in ReorderBufferForget().
3077 *
3078 * This function purges this transaction and its contents from memory and
3079 * disk.
3080 */
3081void
3083 TimestampTz abort_time)
3084{
3085 ReorderBufferTXN *txn;
3086
3088 false);
3089
3090 /* unknown, nothing to remove */
3092 return;
3093
3094 txn->abort_time = abort_time;
3095
3096 /* For streamed transactions notify the remote node about the abort. */
3098 {
3100
3101 /*
3102 * We might have decoded changes for this transaction that could load
3103 * the cache as per the current transaction's view (consider DDL's
3104 * happened in this transaction). We don't want the decoding of future
3105 * transactions to use those cache entries so execute only the inval
3106 * messages in this transaction.
3107 */
3110 txn->invalidations);
3111 }
3112
3113 /* cosmetic... */
3114 txn->final_lsn = lsn;
3115
3116 /* remove potential on-disk data, and deallocate */
3118}
3119
3120/*
3121 * Abort all transactions that aren't actually running anymore because the
3122 * server restarted.
3123 *
3124 * NB: These really have to be transactions that have aborted due to a server
3125 * crash/immediate restart, as we don't deal with invalidations here.
3126 */
3127void
3129{
3131
3132 /*
3133 * Iterate through all (potential) toplevel TXNs and abort all that are
3134 * older than what possibly can be running. Once we've found the first
3135 * that is alive we stop, there might be some that acquired an xid earlier
3136 * but started writing later, but it's unlikely and they will be cleaned
3137 * up in a later call to this function.
3138 */
3140 {
3141 ReorderBufferTXN *txn;
3142
3144
3146 {
3148
3149 /* Notify the remote node about the crash/immediate restart. */
3152
3153 /* remove potential on-disk data, and deallocate this tx */
3155 }
3156 else
3157 return;
3158 }
3159}
3160
3161/*
3162 * Forget the contents of a transaction if we aren't interested in its
3163 * contents. Needs to be first called for subtransactions and then for the
3164 * toplevel xid.
3165 *
3166 * This is significantly different to ReorderBufferAbort() because
3167 * transactions that have committed need to be treated differently from aborted
3168 * ones since they may have modified the catalog.
3169 *
3170 * Note that this is only allowed to be called in the moment a transaction
3171 * commit has just been read, not earlier; otherwise later records referring
3172 * to this xid might re-create the transaction incompletely.
3173 */
3174void
3176{
3177 ReorderBufferTXN *txn;
3178
3180 false);
3181
3182 /* unknown, nothing to forget */
3184 return;
3185
3186 /* this transaction mustn't be streamed */
3188
3189 /* cosmetic... */
3190 txn->final_lsn = lsn;
3191
3192 /*
3193 * Process only cache invalidation messages in this transaction if there
3194 * are any. Even if we're not interested in the transaction's contents, it
3195 * could have manipulated the catalog and we need to update the caches
3196 * according to that.
3197 */
3200 txn->invalidations);
3201 else
3203
3204 /* remove potential on-disk data, and deallocate */
3206}
3207
3208/*
3209 * Invalidate cache for those transactions that need to be skipped just in case
3210 * catalogs were manipulated as part of the transaction.
3211 *
3212 * Note that this is a special-purpose function for prepared transactions where
3213 * we don't want to clean up the TXN even when we decide to skip it. See
3214 * DecodePrepare.
3215 */
3216void
3218{
3219 ReorderBufferTXN *txn;
3220
3222 false);
3223
3224 /* unknown, nothing to do */
3226 return;
3227
3228 /*
3229 * Process cache invalidation messages if there are any. Even if we're not
3230 * interested in the transaction's contents, it could have manipulated the
3231 * catalog and we need to update the caches according to that.
3232 */
3235 txn->invalidations);
3236 else
3238}
3239
3240
3241/*
3242 * Execute invalidations happening outside the context of a decoded
3243 * transaction. That currently happens either for xid-less commits
3244 * (cf. RecordTransactionCommit()) or for invalidations in uninteresting
3245 * transactions (via ReorderBufferForget()).
3246 */
3247void
3249 SharedInvalidationMessage *invalidations)
3250{
3254
3257
3258 /*
3259 * Force invalidations to happen outside of a valid transaction - that way
3260 * entries will just be marked as invalid without accessing the catalog.
3261 * That's advantageous because we don't need to setup the full state
3262 * necessary for catalog access.
3263 */
3265 AbortCurrentTransaction();
3266
3269
3271 {
3274 CurrentResourceOwner = cowner;
3275 }
3276}
3277
3278/*
3279 * Tell reorderbuffer about an xid seen in the WAL stream. Has to be called at
3280 * least once for every xid in XLogRecord->xl_xid (other places in records
3281 * may, but do not have to be passed through here).
3282 *
3283 * Reorderbuffer keeps some data structures about transactions in LSN order,
3284 * for efficiency. To do that it has to know about when transactions are seen
3285 * first in the WAL. As many types of records are not actually interesting for
3286 * logical decoding, they do not necessarily pass through here.
3287 */
3288void
3290{
3291 /* many records won't have an xid assigned, centralize check here */
3294}
3295
3296/*
3297 * Add a new snapshot to this transaction that may only used after lsn 'lsn'
3298 * because the previous snapshot doesn't describe the catalog correctly for
3299 * following rows.
3300 */
3301void
3304{
3306
3309
3311}
3312
3313/*
3314 * Set up the transaction's base snapshot.
3315 *
3316 * If we know that xid is a subtransaction, set the base snapshot on the
3317 * top-level transaction instead.
3318 */
3319void
3322{
3323 ReorderBufferTXN *txn;
3325
3327
3328 /*
3329 * Fetch the transaction to operate on. If we know it's a subtransaction,
3330 * operate on its top-level transaction instead.
3331 */
3337
3339 txn->base_snapshot_lsn = lsn;
3341
3343}
3344
3345/*
3346 * Access the catalog with this CommandId at this point in the changestream.
3347 *
3348 * May only be called for command ids > 1
3349 */
3350void
3353{
3355
3358
3360}
3361
3362/*
3363 * Update memory counters to account for the new or removed change.
3364 *
3365 * We update two counters - in the reorder buffer, and in the transaction
3366 * containing the change. The reorder buffer counter allows us to quickly
3367 * decide if we reached the memory limit, the transaction counter allows
3368 * us to quickly pick the largest transaction for eviction.
3369 *
3370 * Either txn or change must be non-NULL at least. We update the memory
3371 * counter of txn if it's non-NULL, otherwise change->txn.
3372 *
3373 * When streaming is enabled, we need to update the toplevel transaction
3374 * counters instead - we don't really care about subtransactions as we
3375 * can't stream them individually anyway, and we only pick toplevel
3376 * transactions for eviction. So only toplevel transactions matter.
3377 */
3378static void
3380 ReorderBufferChange *change,
3381 ReorderBufferTXN *txn,
3383{
3384 ReorderBufferTXN *toptxn;
3385
3386 Assert(txn || change);
3387
3388 /*
3389 * Ignore tuple CID changes, because those are not evicted when reaching
3390 * memory limit. So we just don't count them, because it might easily
3391 * trigger a pointless attempt to spill.
3392 */
3394 return;
3395
3397 return;
3398
3400 txn = change->txn;
3402
3403 /*
3404 * Update the total size in top level as well. This is later used to
3405 * compute the decoding stats.
3406 */
3407 toptxn = rbtxn_get_toptxn(txn);
3408
3410 {
3412
3415
3416 /* Update the total size in the top transaction. */
3418
3419 /* Update the max-heap */
3423 }
3424 else
3425 {
3429
3430 /* Update the total size in the top transaction. */
3432
3433 /* Update the max-heap */
3437 }
3438
3440}
3441
3442/*
3443 * Add new (relfilelocator, tid) -> (cmin, cmax) mappings.
3444 *
3445 * We do not include this change type in memory accounting, because we
3446 * keep CIDs in a separate list and do not evict them when reaching
3447 * the memory limit.
3448 */
3449void
3454{
3456 ReorderBufferTXN *txn;
3457
3459
3465 change->lsn = lsn;
3466 change->txn = txn;
3468
3470 txn->ntuplecids++;
3471}
3472
3473/*
3474 * Add new invalidation messages to the reorder buffer queue.
3475 */
3476static void
3479 SharedInvalidationMessage *msgs)
3480{
3481 ReorderBufferChange *change;
3482
3489
3491}
3492
3493/*
3494 * A helper function for ReorderBufferAddInvalidations() and
3495 * ReorderBufferAddDistributedInvalidations() to accumulate the invalidation
3496 * messages to the **invals_out.
3497 */
3498static void
3503{
3505 {
3509 }
3510 else
3511 {
3512 /* Enlarge the array of inval messages */
3513 *invals_out =
3519 }
3520}
3521
3522/*
3523 * Accumulate the invalidations for executing them later.
3524 *
3525 * This needs to be called for each XLOG_XACT_INVALIDATIONS message and
3526 * accumulates all the invalidation messages in the toplevel transaction, if
3527 * available, otherwise in the current transaction, as well as in the form of
3528 * change in reorder buffer. We require to record it in form of the change
3529 * so that we can execute only the required invalidations instead of executing
3530 * all the invalidations on each CommandId increment. We also need to
3531 * accumulate these in the txn buffer because in some cases where we skip
3532 * processing the transaction (see ReorderBufferForget), we need to execute
3533 * all the invalidations together.
3534 */
3535void
3538 SharedInvalidationMessage *msgs)
3539{
3540 ReorderBufferTXN *txn;
3541 MemoryContext oldcontext;
3542
3544
3546
3547 /*
3548 * Collect all the invalidations under the top transaction, if available,
3549 * so that we can execute them all together. See comments atop this
3550 * function.
3551 */
3552 txn = rbtxn_get_toptxn(txn);
3553
3554 Assert(nmsgs > 0);
3555
3557 &txn->ninvalidations,
3558 msgs, nmsgs);
3559
3561
3562 MemoryContextSwitchTo(oldcontext);
3563}
3564
3565/*
3566 * Accumulate the invalidations distributed by other committed transactions
3567 * for executing them later.
3568 *
3569 * This function is similar to ReorderBufferAddInvalidations() but stores
3570 * the given inval messages to the txn->invalidations_distributed with the
3571 * overflow check.
3572 *
3573 * This needs to be called by committed transactions to distribute their
3574 * inval messages to in-progress transactions.
3575 */
3576void
3579 SharedInvalidationMessage *msgs)
3580{
3581 ReorderBufferTXN *txn;
3582 MemoryContext oldcontext;
3583
3585
3587
3588 /*
3589 * Collect all the invalidations under the top transaction, if available,
3590 * so that we can execute them all together. See comments
3591 * ReorderBufferAddInvalidations.
3592 */
3593 txn = rbtxn_get_toptxn(txn);
3594
3595 Assert(nmsgs > 0);
3596
3598 {
3599 /*
3600 * Check the transaction has enough space for storing distributed
3601 * invalidation messages.
3602 */
3604 {
3605 /*
3606 * Mark the invalidation message as overflowed and free up the
3607 * messages accumulated so far.
3608 */
3610
3612 {
3615 txn->ninvalidations_distributed = 0;
3616 }
3617 }
3618 else
3620 &txn->ninvalidations_distributed,
3621 msgs, nmsgs);
3622 }
3623
3624 /* Queue the invalidation messages into the transaction */
3626
3627 MemoryContextSwitchTo(oldcontext);
3628}
3629
3630/*
3631 * Apply all invalidations we know. Possibly we only need parts at this point
3632 * in the changestream but we don't know which those are.
3633 */
3634static void
3636{
3639}
3640
3641/*
3642 * Mark a transaction as containing catalog changes
3643 */
3644void
3646 XLogRecPtr lsn)
3647{
3648 ReorderBufferTXN *txn;
3649
3651
3653 {
3656 }
3657
3658 /*
3659 * Mark top-level transaction as having catalog changes too if one of its
3660 * children has so that the ReorderBufferBuildTupleCidHash can
3661 * conveniently check just top-level transaction and decide whether to
3662 * build the hash table or not.
3663 */
3665 {
3667
3669 {
3672 }
3673 }
3674}
3675
3676/*
3677 * Return palloc'ed array of the transactions that have changed catalogs.
3678 * The returned array is sorted in xidComparator order.
3679 *
3680 * The caller must free the returned array when done with it.
3681 */
3682TransactionId *
3684{
3685 dlist_iter iter;
3687 size_t xcnt = 0;
3688
3689 /* Quick return if the list is empty */
3692
3693 /* Initialize XID array */
3696 {
3698 catchange_node,
3699 iter.cur);
3700
3702
3703 xids[xcnt++] = txn->xid;
3704 }
3705
3707
3709 return xids;
3710}
3711
3712/*
3713 * Query whether a transaction is already *known* to contain catalog
3714 * changes. This can be wrong until directly before the commit!
3715 */
3716bool
3718{
3719 ReorderBufferTXN *txn;
3720
3722 false);
3724 return false;
3725
3727}
3728
3729/*
3730 * ReorderBufferXidHasBaseSnapshot
3731 * Have we already set the base snapshot for the given txn/subtxn?
3732 */
3733bool
3735{
3736 ReorderBufferTXN *txn;
3737
3740
3741 /* transaction isn't known yet, ergo no snapshot */
3743 return false;
3744
3745 /* a known subtxn? operate on top-level txn instead */
3749
3751}
3752
3753
3754/*
3755 * ---------------------------------------
3756 * Disk serialization support
3757 * ---------------------------------------
3758 */
3759
3760/*
3761 * Ensure the IO buffer is >= sz.
3762 */
3763static void
3765{
3767 {
3770 }
3772 {
3775 }
3776}
3777
3778
3779/* Compare two transactions by size */
3780static int
3782{
3785
3787 return -1;
3789 return 1;
3790 return 0;
3791}
3792
3793/*
3794 * Find the largest transaction (toplevel or subxact) to evict (spill to disk).
3795 */
3798{
3800
3801 /* Get the largest transaction from the max-heap */
3804
3808
3810}
3811
3812/*
3813 * Find the largest streamable (and non-aborted) toplevel transaction to evict
3814 * (by streaming).
3815 *
3816 * This can be seen as an optimized version of ReorderBufferLargestTXN, which
3817 * should give us the same transaction (because we don't update memory account
3818 * for subtransaction with streaming, so it's always 0). But we can simply
3819 * iterate over the limited number of toplevel transactions that have a base
3820 * snapshot. There is no use of selecting a transaction that doesn't have base
3821 * snapshot because we don't decode such transactions. Also, we do not select
3822 * the transaction which doesn't have any streamable change.
3823 *
3824 * Note that, we skip transactions that contain incomplete changes. There
3825 * is a scope of optimization here such that we can select the largest
3826 * transaction which has incomplete changes. But that will make the code and
3827 * design quite complex and that might not be worth the benefit. If we plan to
3828 * stream the transactions that contain incomplete changes then we need to
3829 * find a way to partially stream/truncate the transaction changes in-memory
3830 * and build a mechanism to partially truncate the spilled files.
3831 * Additionally, whenever we partially stream the transaction we need to
3832 * maintain the last streamed lsn and next time we need to restore from that
3833 * segment and the offset in WAL. As we stream the changes from the top
3834 * transaction and restore them subtransaction wise, we need to even remember
3835 * the subxact from where we streamed the last change.
3836 */
3839{
3840 dlist_iter iter;
3843
3844 /* Find the largest top-level transaction having a base snapshot. */
3846 {
3847 ReorderBufferTXN *txn;
3848
3850
3851 /* must not be a subtxn */
3853 /* base_snapshot must be set */
3855
3856 /* Don't consider these kinds of transactions for eviction. */
3858 !rbtxn_has_streamable_change(txn) ||
3859 rbtxn_is_aborted(txn))
3860 continue;
3861
3862 /* Find the largest of the eviction candidates. */
3864 (txn->total_size > 0))
3865 {
3866 largest = txn;
3868 }
3869 }
3870
3872}
3873
3874/*
3875 * Check whether the logical_decoding_work_mem limit was reached, and if yes
3876 * pick the largest (sub)transaction at-a-time to evict and spill its changes to
3877 * disk or send to the output plugin until we reach under the memory limit.
3878 *
3879 * If debug_logical_replication_streaming is set to "immediate", stream or
3880 * serialize the changes immediately.
3881 *
3882 * XXX At this point we select the transactions until we reach under the memory
3883 * limit, but we might also adapt a more elaborate eviction strategy - for example
3884 * evicting enough transactions to free certain fraction (e.g. 50%) of the memory
3885 * limit.
3886 */
3887static void
3889{
3890 ReorderBufferTXN *txn;
3892
3894 {
3895 /*
3896 * Update the statistics as the memory usage has reached the limit. We
3897 * report the statistics update later in this function since we can
3898 * update the slot statistics altogether while streaming or
3899 * serializing transactions in most cases.
3900 */
3901 rb->memExceededCount += 1;
3902 }
3904 {
3905 /*
3906 * Bail out if debug_logical_replication_streaming is buffered and we
3907 * haven't exceeded the memory limit.
3908 */
3909 return;
3910 }
3911
3912 /*
3913 * If debug_logical_replication_streaming is immediate, loop until there's
3914 * no change. Otherwise, loop until we reach under the memory limit. One
3915 * might think that just by evicting the largest (sub)transaction we will
3916 * come under the memory limit based on assumption that the selected
3917 * transaction is at least as large as the most recent change (which
3918 * caused us to go over the memory limit). However, that is not true
3919 * because a user can reduce the logical_decoding_work_mem to a smaller
3920 * value before the most recent change.
3921 */
3924 rb->size > 0))
3925 {
3926 /*
3927 * Pick the largest non-aborted transaction and evict it from memory
3928 * by streaming, if possible. Otherwise, spill to disk.
3929 */
3932 {
3933 /* we know there has to be one, because the size is not zero */
3937
3938 /* skip the transaction if aborted */
3940 continue;
3941
3943 }
3944 else
3945 {
3946 /*
3947 * Pick the largest transaction (or subtransaction) and evict it
3948 * from memory by serializing it to disk.
3949 */
3951
3952 /* we know there has to be one, because the size is not zero */
3953 Assert(txn);
3956
3957 /* skip the transaction if aborted */
3959 continue;
3960
3962 }
3963
3964 /*
3965 * After eviction, the transaction should have no entries in memory,
3966 * and should use 0 bytes for changes.
3967 */
3970
3971 /*
3972 * We've reported the memExceededCount update while streaming or
3973 * serializing the transaction.
3974 */
3976 }
3977
3980
3981 /* We must be under the memory limit now. */
3983}
3984
3985/*
3986 * Spill data of a large transaction (and its subtransactions) to disk.
3987 */
3988static void
3990{
3995 Size spilled = 0;
3997
4000
4001 /* do the same to all child TXs */
4003 {
4005
4008 }
4009
4010 /* serialize changestream */
4012 {
4013 ReorderBufferChange *change;
4014
4016
4017 /*
4018 * store in segment in which it belongs by start lsn, don't split over
4019 * multiple segments tho
4020 */
4023 {
4025
4028
4030
4031 /*
4032 * No need to care about TLIs here, only used during a single run,
4033 * so each LSN only maps to a specific WAL record.
4034 */
4036 curOpenSegNo);
4037
4038 /* open segment, create it if necessary */
4041
4044 (errcode_for_file_access(),
4046 }
4047
4051
4052 spilled++;
4053 }
4054
4055 /* Update the memory counter */
4057
4058 /* update the statistics iff we have spilled anything */
4059 if (spilled)
4060 {
4061 rb->spillCount += 1;
4062 rb->spillBytes += size;
4063
4064 /* don't consider already serialized transactions */
4066
4067 /* update the decoding stats */
4069 }
4070
4073 txn->nentries_mem = 0;
4075
4078}
4079
4080/*
4081 * Serialize individual change to disk.
4082 */
4083static void
4086{
4087 ReorderBufferDiskChange *ondisk;
4089
4091
4094
4096 {
4097 /* fall through these, they're all similar enough */
4102 {
4105 newtup;
4108
4111
4113 {
4117 }
4118
4120 {
4124 }
4125
4126 /* make sure we have enough space */
4128
4130 /* might have been reallocated above */
4132
4134 {
4137
4140 }
4141
4143 {
4146
4149 }
4150 break;
4151 }
4153 {
4156
4160
4162
4163 /* might have been reallocated above */
4165
4166 /* write the prefix including the size */
4170 prefix_size);
4171 data += prefix_size;
4172
4173 /* write the message including the size */
4179
4180 break;
4181 }
4183 {
4187
4189
4192
4193 /* might have been reallocated above */
4197
4198 break;
4199 }
4201 {
4204
4206
4210
4211 /* make sure we have enough space */
4214 /* might have been reallocated above */
4216
4219
4221 {
4225 }
4226
4228 {
4232 }
4233 break;
4234 }
4236 {
4237 Size size;
4239
4240 /* account for the OIDs of truncated relations */
4242 sz += size;
4243
4244 /* make sure we have enough space */
4246
4248 /* might have been reallocated above */
4250
4252 data += size;
4253
4254 break;
4255 }
4260 /* ReorderBufferChange contains everything important */
4261 break;
4262 }
4263
4264 ondisk->size = sz;
4265
4266 errno = 0;
4269 {
4271
4273
4274 /* if write didn't set errno, assume problem is no disk space */
4277 (errcode_for_file_access(),
4279 txn->xid)));
4280 }
4281 pgstat_report_wait_end();
4282
4283 /*
4284 * Keep the transaction's final_lsn up to date with each change we send to
4285 * disk, so that ReorderBufferRestoreCleanup works correctly. (We used to
4286 * only do this on commit and abort records, but that doesn't work if a
4287 * system crash leaves a transaction without its abort record).
4288 *
4289 * Make sure not to move it backwards.
4290 */
4293
4295}
4296
4297/* Returns true, if the output plugin supports streaming, false, otherwise. */
4298static inline bool
4300{
4302
4304}
4305
4306/* Returns true, if the streaming can be started now, false, otherwise. */
4307static inline bool
4309{
4312
4313 /* We can't start streaming unless a consistent state is reached. */
4315 return false;
4316
4317 /*
4318 * We can't start streaming immediately even if the streaming is enabled
4319 * because we previously decoded this transaction and now just are
4320 * restarting.
4321 */
4324 return true;
4325
4326 return false;
4327}
4328
4329/*
4330 * Send data of a large transaction (and its subtransactions) to the
4331 * output plugin, but using the stream API.
4332 */
4333static void
4335{
4336 Snapshot snapshot_now;
4337 CommandId command_id;
4338 Size stream_bytes;
4340
4341 /* We can never reach here for a subtransaction. */
4343
4344 /*
4345 * We can't make any assumptions about base snapshot here, similar to what
4346 * ReorderBufferCommit() does. That relies on base_snapshot getting
4347 * transferred from subxact in ReorderBufferCommitChild(), but that was
4348 * not yet called as the transaction is in-progress.
4349 *
4350 * So just walk the subxacts and use the same logic here. But we only need
4351 * to do that once, when the transaction is streamed for the first time.
4352 * After that we need to reuse the snapshot from the previous run.
4353 *
4354 * Unlike DecodeCommit which adds xids of all the subtransactions in
4355 * snapshot's xip array via SnapBuildCommitTxn, we can't do that here but
4356 * we do add them to subxip array instead via ReorderBufferCopySnap. This
4357 * allows the catalog changes made in subtransactions decoded till now to
4358 * be visible.
4359 */
4361 {
4363
4364 /* make sure this transaction is streamed for the first time */
4366
4367 /* at the beginning we should have invalid command ID */
4369
4371 {
4373
4376 }
4377
4378 /*
4379 * If this transaction has no snapshot, it didn't make any changes to
4380 * the database till now, so there's nothing to decode.
4381 */
4383 {
4385 return;
4386 }
4387
4388 command_id = FirstCommandId;
4390 txn, command_id);
4391 }
4392 else
4393 {
4394 /* the transaction must have been already streamed */
4396
4397 /*
4398 * Nah, we already have snapshot from the previous streaming run. We
4399 * assume new subxacts can't move the LSN backwards, and so can't beat
4400 * the LSN condition in the previous branch (so no need to walk
4401 * through subxacts again). In fact, we must not do that as we may be
4402 * using snapshot half-way through the subxact.
4403 */
4404 command_id = txn->command_id;
4405
4406 /*
4407 * We can't use txn->snapshot_now directly because after the last
4408 * streaming run, we might have got some new sub-transactions. So we
4409 * need to add them to the snapshot.
4410 */
4412 txn, command_id);
4413
4414 /* Free the previously copied snapshot. */
4418 }
4419
4420 /*
4421 * Remember this information to be used later to update stats. We can't
4422 * update the stats here as an error while processing the changes would
4423 * lead to the accumulation of stats even though we haven't streamed all
4424 * the changes.
4425 */
4427 stream_bytes = txn->total_size;
4428
4429 /* Process and send the changes to output plugin. */
4431 command_id, true);
4432
4433 rb->streamCount += 1;
4434 rb->streamBytes += stream_bytes;
4435
4436 /* Don't consider already streamed transaction. */
4438
4439 /* update the decoding stats */
4441
4445}
4446
4447/*
4448 * Size of a change in memory.
4449 */
4452{
4454
4456 {
4457 /* fall through these, they're all similar enough */
4462 {
4464 newtup;
4467
4470
4472 {
4476 }
4477
4479 {
4483 }
4484
4485 break;
4486 }
4488 {
4490
4493
4494 break;
4495 }
4497 {
4500 break;
4501 }
4503 {
4505
4507
4511
4512 break;
4513 }
4515 {
4517
4518 break;
4519 }
4524 /* ReorderBufferChange contains everything important */
4525 break;
4526 }
4527
4529}
4530
4531
4532/*
4533 * Restore a number of changes spilled to disk back into memory.
4534 */
4538{
4543
4546
4547 /* free current entries, so we have memory for more */
4549 {
4552
4555 }
4556 txn->nentries_mem = 0;
4558
4560
4562 {
4564 ReorderBufferDiskChange *ondisk;
4565
4566 CHECK_FOR_INTERRUPTS();
4567
4569 {
4571
4572 /* first time in */
4573 if (*segno == 0)
4575
4577
4578 /*
4579 * No need to care about TLIs here, only used during a single run,
4580 * so each LSN only maps to a specific WAL record.
4581 */
4583 *segno);
4584
4586
4587 /* No harm in resetting the offset even in case of failure */
4588 file->curOffset = 0;
4589
4591 {
4592 *fd = -1;
4593 (*segno)++;
4594 continue;
4595 }
4598 (errcode_for_file_access(),
4600 path)));
4601 }
4602
4603 /*
4604 * Read the statically sized part of a change which has information
4605 * about the total size. If we couldn't read a record, we're at the
4606 * end of this file.
4607 */
4612
4613 /* eof */
4615 {
4617 *fd = -1;
4618 (*segno)++;
4619 continue;
4620 }
4623 (errcode_for_file_access(),
4627 (errcode_for_file_access(),
4629 readBytes,
4631
4633
4635
4639
4643 file->curOffset,
4645
4648 (errcode_for_file_access(),
4652 (errcode_for_file_access(),
4654 readBytes,
4656
4658
4659 /*
4660 * ok, read a full change from disk, now restore it into proper
4661 * in-memory format
4662 */
4664 restored++;
4665 }
4666
4668}
4669
4670/*
4671 * Convert change from its on-disk format to in-memory format and queue it onto
4672 * the TXN's ->changes list.
4673 *
4674 * Note: although "data" is declared char*, at entry it points to a
4675 * maxalign'd buffer, making it safe in most of this function to assume
4676 * that the pointed-to data is suitably aligned for direct access.
4677 */
4678static void
4681{
4682 ReorderBufferDiskChange *ondisk;
4683 ReorderBufferChange *change;
4684
4686
4688
4689 /* copy static part */
4691
4693
4694 /* restore individual stuff */
4696 {
4697 /* fall through these, they're all similar enough */
4703 {
4705
4708
4709 /* restore ->tuple */
4713
4714 /* reset t_data pointer into the new tuplebuf */
4717
4718 /* restore tuple data itself */
4721 }
4722
4724 {
4725 /* here, data might not be suitably aligned! */
4727
4730
4733
4734 /* restore ->tuple */
4738
4739 /* reset t_data pointer into the new tuplebuf */
4742
4743 /* restore tuple data itself */
4746 }
4747
4748 break;
4750 {
4751 Size prefix_size;
4752
4753 /* read prefix */
4757 prefix_size);
4760 data += prefix_size;
4761
4762 /* read the message */
4770
4771 break;
4772 }
4774 {
4777
4780
4781 /* read the message */
4783
4784 break;
4785 }
4787 {
4790 Size size;
4791
4793
4797
4799
4801
4807 break;
4808 }
4809 /* the base struct contains all the data, easy peasy */
4811 {
4812 Oid *relids;
4813
4817
4818 break;
4819 }
4824 break;
4825 }
4826
4828 txn->nentries_mem++;
4829
4830 /*
4831 * Update memory accounting for the restored change. We need to do this
4832 * although we don't check the memory limit when restoring the changes in
4833 * this branch (we only do that when initially queueing the changes after
4834 * decoding), because we will release the changes later, and that will
4835 * update the accounting too (subtracting the size from the counters). And
4836 * we don't want to underflow there.
4837 */
4839 ReorderBufferChangeSize(change));
4840}
4841
4842/*
4843 * Remove all on-disk stored for the passed in transaction.
4844 */
4845static void
4847{
4848 XLogSegNo first;
4850 XLogSegNo last;
4851
4854
4857
4858 /* iterate over all possible filenames, and delete them */
4860 {
4862
4866 (errcode_for_file_access(),
4868 }
4869}
4870
4871/*
4872 * Remove any leftover serialized reorder buffers from a slot directory after a
4873 * prior crash or decoding session exit.
4874 */
4875static void
4877{
4882
4884
4885 /* we're only handling directories here, skip if it's not ours */
4887 return;
4888
4891 {
4892 /* only look at names that can be ours */
4894 {
4897 spill_de->d_name);
4898
4901 (errcode_for_file_access(),
4903 path, PG_REPLSLOT_DIR, slotname)));
4904 }
4905 }
4907}
4908
4909/*
4910 * Given a replication slot, transaction ID and segment number, fill in the
4911 * corresponding spill file into 'path', which is a caller-owned buffer of size
4912 * at least MAXPGPATH.
4913 */
4914static void
4916 XLogSegNo segno)
4917{
4919
4921
4923 PG_REPLSLOT_DIR,
4926}
4927
4928/*
4929 * Delete all data spilled to disk after we've restarted/crashed. It will be
4930 * recreated when the respective slots are reused.
4931 */
4932void
4934{
4937
4940 {
4943 continue;
4944
4945 /* if it cannot be a slot, skip the directory */
4947 continue;
4948
4949 /*
4950 * ok, has to be a surviving logical slot, iterate and delete
4951 * everything starting with xid-*
4952 */
4954 }
4956}
4957
4958/* ---------------------------------------
4959 * toast reassembly support
4960 * ---------------------------------------
4961 */
4962
4963/*
4964 * Initialize per tuple toast reconstruction support.
4965 */
4966static void
4968{
4970
4972
4978}
4979
4980/*
4981 * Per toast-chunk handling for toast reconstruction
4982 *
4983 * Appends a toast chunk so we can reconstruct it when the tuple "owning" the
4984 * toasted Datum comes along.
4985 */
4986static void
4989{
4992 bool found;
4994 bool isnull;
4995 Pointer chunk;
4997 Oid chunk_id;
4999
5002
5004
5007 Assert(!isnull);
5009 Assert(!isnull);
5010
5013
5014 if (!found)
5015 {
5017 ent->num_chunks = 0;
5018 ent->last_chunk_seq = 0;
5019 ent->size = 0;
5022
5025 chunk_seq, chunk_id);
5026 }
5030
5032 Assert(!isnull);
5033
5034 /* calculate size so we can allocate the right size at once later */
5038 /* could happen due to heap_form_tuple doing its thing */
5040 else
5042
5045 ent->num_chunks++;
5047}
5048
5049/*
5050 * Rejigger change->newtuple to point to in-memory toast tuples instead of
5051 * on-disk toast tuples that may no longer exist (think DROP TABLE or VACUUM).
5052 *
5053 * We cannot replace unchanged toast tuples though, so those will still point
5054 * to on-disk toast data.
5055 *
5056 * While updating the existing change with detoasted tuple data, we need to
5057 * update the memory accounting info, because the change size will differ.
5058 * Otherwise the accounting may get out of sync, triggering serialization
5059 * at unexpected times.
5060 *
5061 * We simply subtract size of the change before rejiggering the tuple, and
5062 * then add the new size. This makes it look like the change was removed
5063 * and then added back, except it only tweaks the accounting info.
5064 *
5065 * In particular it can't trigger serialization, which would be pointless
5066 * anyway as it happens during commit processing right before handing
5067 * the change to the output plugin.
5068 */
5069static void
5072{
5073 TupleDesc desc;
5076 bool *isnull;
5079 Relation toast_rel;
5081 MemoryContext oldcontext;
5084
5085 /* no toast tuples changed */
5087 return;
5088
5089 /*
5090 * We're going to modify the size of the change. So, to make sure the
5091 * accounting is correct we record the current change size and then after
5092 * re-computing the change we'll subtract the recorded size and then
5093 * re-add the new change size at the end. We don't immediately subtract
5094 * the old size because if there is any error before we add the new size,
5095 * we will release the changes and that will update the accounting info
5096 * (subtracting the size from the counters). And we don't want to
5097 * underflow there.
5098 */
5100
5102
5103 /* we should only have toast tuples in an INSERT or UPDATE */
5105
5106 desc = RelationGetDescr(relation);
5107
5112
5114
5115 /* should we allocate from stack instead? */
5119
5121
5123
5125 {
5129
5130 /* va_rawsize is the size of the original datum -- including header */
5131 varatt_external toast_pointer;
5134 varlena *reconstructed;
5137
5139 continue;
5140
5141 /* not a varlena datatype */
5143 continue;
5144
5145 /* no data */
5147 continue;
5148
5149 /* ok, we know we have a toast datum */
5151
5152 /* no need to do anything if the tuple isn't external */
5154 continue;
5155
5157
5158 /*
5159 * Check whether the toast tuple changed, replace if so.
5160 */
5163 &toast_pointer.va_valueid,
5164 HASH_FIND,
5165 NULL);
5167 continue;
5168
5169 new_datum =
5171
5173
5175
5176 ent->reconstructed = reconstructed;
5177
5178 /* stitch toast tuple back together from its parts */
5180 {
5184 Pointer chunk;
5185
5189
5193
5195 VARDATA(chunk),
5198 }
5200
5201 /* make sure its marked as compressed or not */
5204 else
5206
5208 redirect_pointer.pointer = reconstructed;
5209
5213
5215 }
5216
5217 /*
5218 * Build tuple in separate memory & copy tuple back into the tuplebuf
5219 * passed to the output plugin. We can't directly heap_fill_tuple() into
5220 * the tuplebuf because attrs[] will point back into the current content.
5221 */
5225
5228
5229 /*
5230 * free resources we won't further need, more persistent stuff will be
5231 * free'd in ReorderBufferToastReset().
5232 */
5233 RelationClose(toast_rel);
5236 {
5239 }
5242 pfree(isnull);
5243
5244 MemoryContextSwitchTo(oldcontext);
5245
5246 /* subtract the old change size */
5248 /* now add the change back, with the correct size */
5250 ReorderBufferChangeSize(change));
5251}
5252
5253/*
5254 * Free all resources allocated for toast reconstruction.
5255 */
5256static void
5258{
5261
5263 return;
5264
5265 /* sequentially walk over the hash and free everything */
5268 {
5270
5273
5275 {
5276 ReorderBufferChange *change =
5278
5281 }
5282 }
5283
5286}
5287
5288
5289/* ---------------------------------------
5290 * Visibility support for logical decoding
5291 *
5292 *
5293 * Lookup actual cmin/cmax values when using decoding snapshot. We can't
5294 * always rely on stored cmin/cmax values because of two scenarios:
5295 *
5296 * * A tuple got changed multiple times during a single transaction and thus
5297 * has got a combo CID. Combo CIDs are only valid for the duration of a
5298 * single transaction.
5299 * * A tuple with a cmin but no cmax (and thus no combo CID) got
5300 * deleted/updated in another transaction than the one which created it
5301 * which we are looking at right now. As only one of cmin, cmax or combo CID
5302 * is actually stored in the heap we don't have access to the value we
5303 * need anymore.
5304 *
5305 * To resolve those problems we have a per-transaction hash of (cmin,
5306 * cmax) tuples keyed by (relfilelocator, ctid) which contains the actual
5307 * (cmin, cmax) values. That also takes care of combo CIDs by simply
5308 * not caring about them at all. As we have the real cmin/cmax values
5309 * combo CIDs aren't interesting.
5310 *
5311 * As we only care about catalog tuples here the overhead of this
5312 * hashtable should be acceptable.
5313 *
5314 * Heap rewrites complicate this a bit, check rewriteheap.c for
5315 * details.
5316 * -------------------------------------------------------------------------
5317 */
5318
5319/* struct for sorting mapping files by LSN efficiently */
5321{
5324} RewriteMappingFile;
5325
5326#ifdef NOT_USED
5327static void
5329{
5332
5335 {
5337 ent->key.rlocator.dbOid,
5338 ent->key.rlocator.spcOid,
5339 ent->key.rlocator.relNumber,
5342 ent->cmin,
5343 ent->cmax
5344 );
5345 }
5346}
5347#endif
5348
5349/*
5350 * Apply a single mapping file to tuplecid_data.
5351 *
5352 * The mapping file has to have been verified to be a) committed b) for our
5353 * transaction c) applied in LSN order.
5354 */
5355static void
5357{
5361 LogicalRewriteMappingData map;
5362
5367 (errcode_for_file_access(),
5369
5370 while (true)
5371 {
5375 bool found;
5376
5377 /* be careful about padding */
5379
5380 /* read all mappings till the end of the file */
5383 pgstat_report_wait_end();
5384
5387 (errcode_for_file_access(),
5389 path)));
5391 break;
5394 (errcode_for_file_access(),
5396 path, readBytes,
5398
5401 &key.tid);
5402
5403
5406
5407 /* no existing mapping, no need to update */
5409 continue;
5410
5413 &key.tid);
5414
5417
5418 if (found)
5419 {
5420 /*
5421 * Make sure the existing mapping makes sense. We sometime update
5422 * old records that did not yet have a cmax (e.g. pg_class' own
5423 * entry while rewriting it) during rewrites, so allow that.
5424 */
5427 }
5428 else
5429 {
5430 /* update mapping */
5434 }
5435 }
5436
5439 (errcode_for_file_access(),
5441}
5442
5443
5444/*
5445 * Check whether the TransactionId 'xid' is in the pre-sorted array 'xip'.
5446 */
5447static bool
5449{
5452}
5453
5454/*
5455 * list_sort() comparator for sorting RewriteMappingFiles in LSN order.
5456 */
5457static int
5459{
5462
5464}
5465
5466/*
5467 * Apply any existing logical remapping files if there are any targeted at our
5468 * transaction for relid.
5469 */
5470static void
5472{
5476 ListCell *file;
5478
5481 {
5488 f_lo;
5489 RewriteMappingFile *f;
5490
5493 continue;
5494
5495 /* Ignore files that aren't ours */
5497 continue;
5498
5503
5505
5506 /* mapping for another database */
5508 continue;
5509
5510 /* mapping for another relation */
5512 continue;
5513
5514 /* did the creating transaction abort? */
5516 continue;
5517
5518 /* not for our transaction */
5520 continue;
5521
5522 /* ok, relevant, queue for apply */
5526 files = lappend(files, f);
5527 }
5529
5530 /* sort files so we apply them in LSN order */
5532
5533 foreach(file, files)
5534 {
5536
5538 snapshot->subxip[0]);
5540 pfree(f);
5541 }
5542}
5543
5544/*
5545 * Lookup cmin/cmax of a tuple, during logical decoding where we can't rely on
5546 * combo CIDs.
5547 */
5548bool
5550 Snapshot snapshot,
5553{
5556 ForkNumber forkno;
5557 BlockNumber blockno;
5559
5560 /*
5561 * Return unresolved if tuplecid_data is not valid. That's because when
5562 * streaming in-progress transactions we may run into tuples with the CID
5563 * before actually decoding them. Think e.g. about INSERT followed by
5564 * TRUNCATE, where the TRUNCATE may not be decoded yet when applying the
5565 * INSERT. So in such cases, we assume the CID is from the future
5566 * command.
5567 */
5569 return false;
5570
5571 /* be careful about padding */
5573
5575
5576 /*
5577 * get relfilelocator from the buffer, no convenient way to access it
5578 * other than that.
5579 */
5581
5582 /* tuples can only be in the main fork */
5585
5587 &key.tid);
5588
5589restart:
5592
5593 /*
5594 * failed to find a mapping, check whether the table was rewritten and
5595 * apply mapping if so, but only do that once - there can be no new
5596 * mappings while we are in here since we have to hold a lock on the
5597 * relation.
5598 */
5600 {
5602 /* now check but don't update for a mapping again */
5604 goto restart;
5605 }
5607 return false;
5608
5609 if (cmin)
5610 *cmin = ent->cmin;
5611 if (cmax)
5612 *cmax = ent->cmax;
5613 return true;
5614}
5615
5616/*
5617 * Count invalidation messages of specified transaction.
5618 *
5619 * Returns number of messages, and msgs is set to the pointer of the linked
5620 * list for the messages.
5621 */
5622uint32
5624 SharedInvalidationMessage **msgs)
5625{
5626 ReorderBufferTXN *txn;
5627
5629 false);
5630
5632 return 0;
5633
5634 *msgs = txn->invalidations;
5635
5637}
void binaryheap_replace_first(binaryheap *heap, bh_node_type d)
Definition binaryheap.c:253
bh_node_type binaryheap_remove_first(binaryheap *heap)
Definition binaryheap.c:190
void binaryheap_add_unordered(binaryheap *heap, bh_node_type d)
Definition binaryheap.c:114
binaryheap * binaryheap_allocate(int capacity, binaryheap_comparator compare, void *arg)
Definition binaryheap.c:37
void BufferGetTag(Buffer buffer, RelFileLocator *rlocator, ForkNumber *forknum, BlockNumber *blknum)
Definition bufmgr.c:4490
memcpy(sums, checksumBaseOffsets, sizeof(checksumBaseOffsets))
#define VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr)
Definition detoast.h:22
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
Definition dynahash.c:889
HTAB * hash_create(const char *tabname, int64 nelem, const HASHCTL *info, int flags)
Definition dynahash.c:360
struct dirent * ReadDirExtended(DIR *dir, const char *dirname, int elevel)
Definition fd.c:2972
static ssize_t FileRead(File file, void *buffer, size_t amount, pgoff_t offset, uint32 wait_event_info)
Definition fd.h:225
MemoryContext GenerationContextCreate(MemoryContext parent, const char *name, Size minContextSize, Size initBlockSize, Size maxBlockSize)
Definition generation.c:162
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, const Datum *values, const bool *isnull)
Definition heaptuple.c:1025
void heap_deform_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *values, bool *isnull)
Definition heaptuple.c:1254
static Datum fastgetattr(HeapTuple tup, int attnum, TupleDesc tupleDesc, bool *isnull)
Definition htup_details.h:847
static bool dlist_has_next(const dlist_head *head, const dlist_node *node)
Definition ilist.h:503
static void dclist_push_tail(dclist_head *head, dlist_node *node)
Definition ilist.h:709
static void dlist_insert_before(dlist_node *before, dlist_node *node)
Definition ilist.h:393
static dlist_node * dlist_next_node(dlist_head *head, dlist_node *node)
Definition ilist.h:537
static void dlist_push_tail(dlist_head *head, dlist_node *node)
Definition ilist.h:364
static void dclist_delete_from(dclist_head *head, dlist_node *node)
Definition ilist.h:763
void LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
Definition inval.c:823
static OffsetNumber ItemPointerGetOffsetNumber(const ItemPointerData *pointer)
Definition itemptr.h:124
static BlockNumber ItemPointerGetBlockNumber(const ItemPointerData *pointer)
Definition itemptr.h:103
static void ItemPointerCopy(const ItemPointerData *fromPointer, ItemPointerData *toPointer)
Definition itemptr.h:172
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition mcxt.c:1269
void pairingheap_remove(pairingheap *heap, pairingheap_node *node)
Definition pairingheap.c:184
void pairingheap_add(pairingheap *heap, pairingheap_node *node)
Definition pairingheap.c:126
pairingheap * pairingheap_allocate(pairingheap_comparator compare, void *arg)
Definition pairingheap.c:42
pairingheap_node * pairingheap_first(pairingheap *heap)
Definition pairingheap.c:144
#define pairingheap_container(type, membername, ptr)
Definition pairingheap.h:43
#define pairingheap_const_container(type, membername, ptr)
Definition pairingheap.h:51
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition palloc.h:138
bool TransactionIdIsInProgress(TransactionId xid)
Definition procarray.c:1393
Oid RelidByRelfilenumber(Oid reltablespace, RelFileNumber relfilenumber)
Definition relfilenumbermap.c:141
static int file_sort_by_lsn(const ListCell *a_p, const ListCell *b_p)
Definition reorderbuffer.c:5459
void ReorderBufferFreeRelids(ReorderBuffer *rb, Oid *relids)
Definition reorderbuffer.c:642
void ReorderBufferFreeChange(ReorderBuffer *rb, ReorderBufferChange *change, bool upd_mem)
Definition reorderbuffer.c:523
static void ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition reorderbuffer.c:5071
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Definition reorderbuffer.c:3646
static void ReorderBufferStreamCommit(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:1984
void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)
Definition reorderbuffer.c:3352
static void ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:1536
static void ReorderBufferReplay(ReorderBufferTXN *txn, ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, ReplOriginId origin_id, XLogRecPtr origin_lsn)
Definition reorderbuffer.c:2819
static void ReorderBufferAccumulateInvalidations(SharedInvalidationMessage **invals_out, uint32 *ninvals_out, SharedInvalidationMessage *msgs_new, Size nmsgs_new)
Definition reorderbuffer.c:3500
static ReorderBufferTXN * ReorderBufferLargestTXN(ReorderBuffer *rb)
Definition reorderbuffer.c:3798
void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
Definition reorderbuffer.c:3451
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
Definition reorderbuffer.c:3321
static void ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:4847
static void ReorderBufferToastInitHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:4968
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, TimestampTz abort_time)
Definition reorderbuffer.c:3083
static bool ReorderBufferCanStartStreaming(ReorderBuffer *rb)
Definition reorderbuffer.c:4309
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
Definition reorderbuffer.c:3718
void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Definition reorderbuffer.c:3218
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb)
Definition reorderbuffer.c:1073
static int ReorderBufferIterCompare(Datum a, Datum b, void *arg)
Definition reorderbuffer.c:1262
static void ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferIterTXNState *volatile *iter_state)
Definition reorderbuffer.c:1285
bool ResolveCminCmaxDuringDecoding(HTAB *tuplecid_data, Snapshot snapshot, HeapTuple htup, Buffer buffer, CommandId *cmin, CommandId *cmax)
Definition reorderbuffer.c:5550
static void ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition reorderbuffer.c:4988
void ReorderBufferFreeTupleBuf(HeapTuple tuple)
Definition reorderbuffer.c:611
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
Definition reorderbuffer.c:811
void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, char *gid)
Definition reorderbuffer.c:2956
uint32 ReorderBufferGetInvalidations(ReorderBuffer *rb, TransactionId xid, SharedInvalidationMessage **msgs)
Definition reorderbuffer.c:5624
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Definition reorderbuffer.c:3176
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
Definition reorderbuffer.c:1220
TransactionId * ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
Definition reorderbuffer.c:3684
static void ReorderBufferSaveTXNSnapshot(ReorderBuffer *rb, ReorderBufferTXN *txn, Snapshot snapshot_now, CommandId command_id)
Definition reorderbuffer.c:2121
static Size ReorderBufferChangeSize(ReorderBufferChange *change)
Definition reorderbuffer.c:4452
static void AssertChangeLsnOrder(ReorderBufferTXN *txn)
Definition reorderbuffer.c:1014
static bool ReorderBufferCanStream(ReorderBuffer *rb)
Definition reorderbuffer.c:4300
static int ReorderBufferTXNSizeCompare(const pairingheap_node *a, const pairingheap_node *b, void *arg)
Definition reorderbuffer.c:3782
static void ReorderBufferApplyChange(ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change, bool streaming)
Definition reorderbuffer.c:2073
void ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid)
Definition reorderbuffer.c:2935
bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid, XLogRecPtr prepare_lsn, XLogRecPtr end_lsn, TimestampTz prepare_time, ReplOriginId origin_id, XLogRecPtr origin_lsn)
Definition reorderbuffer.c:2903
static void ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, Snapshot snapshot_now, CommandId command_id, XLogRecPtr last_lsn)
Definition reorderbuffer.c:2166
void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, XLogRecPtr two_phase_at, TimestampTz commit_time, ReplOriginId origin_id, XLogRecPtr origin_lsn, char *gid, bool is_commit)
Definition reorderbuffer.c:2997
static void ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn, int fd, ReorderBufferChange *change)
Definition reorderbuffer.c:4085
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
Definition reorderbuffer.c:3537
void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, ReplOriginId origin_id, XLogRecPtr origin_lsn)
Definition reorderbuffer.c:2880
int debug_logical_replication_streaming
Definition reorderbuffer.c:230
void ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
Definition reorderbuffer.c:3578
static void ReorderBufferSerializeReserve(ReorderBuffer *rb, Size sz)
Definition reorderbuffer.c:3765
void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snap, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
Definition reorderbuffer.c:874
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
Definition reorderbuffer.c:3735
static void ReorderBufferExecuteInvalidations(uint32 nmsgs, SharedInvalidationMessage *msgs)
Definition reorderbuffer.c:3636
static void ReorderBufferIterTXNFinish(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
Definition reorderbuffer.c:1505
void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
Definition reorderbuffer.c:3303
static void ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, bool txn_prepared)
Definition reorderbuffer.c:1657
#define CHANGES_THRESHOLD
static ReorderBufferTXN * ReorderBufferLargestStreamableTopTXN(ReorderBuffer *rb)
Definition reorderbuffer.c:3839
static bool ReorderBufferCheckAndTruncateAbortedTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:1775
static void ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn, char *data)
Definition reorderbuffer.c:4680
HeapTuple ReorderBufferAllocTupleBuf(ReorderBuffer *rb, Size tuple_len)
Definition reorderbuffer.c:593
static void ReorderBufferApplyMessage(ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferChange *change, bool streaming)
Definition reorderbuffer.c:2101
static void ReorderBufferFreeSnap(ReorderBuffer *rb, Snapshot snap)
Definition reorderbuffer.c:1969
static void ReorderBufferCleanupSerializedTXNs(const char *slotname)
Definition reorderbuffer.c:4877
ReorderBufferChange * ReorderBufferAllocChange(ReorderBuffer *rb)
Definition reorderbuffer.c:508
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
Definition reorderbuffer.c:1088
static bool TransactionIdInArray(TransactionId xid, TransactionId *xip, Size num)
Definition reorderbuffer.c:5449
static Snapshot ReorderBufferCopySnap(ReorderBuffer *rb, Snapshot orig_snap, ReorderBufferTXN *txn, CommandId cid)
Definition reorderbuffer.c:1910
static void ReorderBufferApplyTruncate(ReorderBuffer *rb, ReorderBufferTXN *txn, int nrelations, Relation *relations, ReorderBufferChange *change, bool streaming)
Definition reorderbuffer.c:2087
static void ReorderBufferProcessPartialChange(ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferChange *change, bool toast_insert)
Definition reorderbuffer.c:742
static void ReorderBufferToastReset(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:5258
static void ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:3990
static void UpdateLogicalMappings(HTAB *tuplecid_data, Oid relid, Snapshot snapshot)
Definition reorderbuffer.c:5472
static void ReorderBufferQueueInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
Definition reorderbuffer.c:3478
static ReorderBufferTXN * ReorderBufferAllocTXN(ReorderBuffer *rb)
Definition reorderbuffer.c:436
static void ReorderBufferFreeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:460
void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
Definition reorderbuffer.c:3249
static void ReorderBufferTransferSnapToParent(ReorderBufferTXN *txn, ReorderBufferTXN *subtxn)
Definition reorderbuffer.c:1166
static void ReorderBufferBuildTupleCidHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:1837
static ReorderBufferChange * ReorderBufferIterTXNNext(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
Definition reorderbuffer.c:1413
Oid * ReorderBufferAllocRelids(ReorderBuffer *rb, int nrelids)
Definition reorderbuffer.c:626
static void ReorderBufferCheckMemoryLimit(ReorderBuffer *rb)
Definition reorderbuffer.c:3889
static void ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb, ReorderBufferChange *change, ReorderBufferTXN *txn, bool addition, Size sz)
Definition reorderbuffer.c:3380
static void ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:4335
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Definition reorderbuffer.c:3290
static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn, TXNEntryFile *file, XLogSegNo *segno)
Definition reorderbuffer.c:4537
void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
Definition reorderbuffer.c:1100
static void ReorderBufferSerializedPath(char *path, ReplicationSlot *slot, TransactionId xid, XLogSegNo segno)
Definition reorderbuffer.c:4916
static void ApplyLogicalMappingFile(HTAB *tuplecid_data, const char *fname)
Definition reorderbuffer.c:5357
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
Definition reorderbuffer.c:3129
static ReorderBufferTXN * ReorderBufferTXNByXid(ReorderBuffer *rb, TransactionId xid, bool create, bool *is_new, XLogRecPtr lsn, bool create_as_top)
Definition reorderbuffer.c:654
static void ReorderBufferMaybeMarkTXNStreamed(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:2139
ReorderBufferTXN * ReorderBufferGetOldestTXN(ReorderBuffer *rb)
Definition reorderbuffer.c:1045
static void ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr commit_lsn, volatile Snapshot snapshot_now, volatile CommandId command_id, bool streaming)
Definition reorderbuffer.c:2204
@ DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE
Definition reorderbuffer.h:34
@ DEBUG_LOGICAL_REP_STREAMING_BUFFERED
Definition reorderbuffer.h:33
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM
Definition reorderbuffer.h:61
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT
Definition reorderbuffer.h:62
@ REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID
Definition reorderbuffer.h:58
@ REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID
Definition reorderbuffer.h:59
@ REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT
Definition reorderbuffer.h:57
MemoryContext SlabContextCreate(MemoryContext parent, const char *name, Size blockSize, Size chunkSize)
Definition slab.c:322
bool ReplicationSlotValidateName(const char *name, bool allow_reserved_name, int elevel)
Definition slot.c:265
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition snapbuild.c:308
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition snapbuild.c:281
void SetupHistoricSnapshot(Snapshot historic_snapshot, HTAB *tuplecids)
Definition snapmgr.c:1668
Definition tupdesc.h:69
Definition dirent.c:26
Definition elog.h:422
Definition hsearch.h:65
Definition hsearch.h:116
Definition dynahash.c:225
Definition htup.h:63
Definition htup_details.h:154
Definition itemptr.h:37
Definition pg_list.h:54
Definition logical.h:34
Definition rewriteheap.h:36
Definition memnodes.h:118
Definition relfilelocator.h:59
Definition rel.h:56
Definition reorderbuffer.h:77
struct ReorderBufferChange::@118::@122 tuplecid
bool clear_toast_afterwards
Definition reorderbuffer.h:101
struct ReorderBufferChange::@118::@120 truncate
struct ReorderBufferChange::@118::@123 inval
struct ReorderBufferChange::@118::@121 msg
SharedInvalidationMessage * invalidations
Definition reorderbuffer.h:155
struct ReorderBufferChange::@118::@119 tp
union ReorderBufferChange::@118 data
Definition reorderbuffer.c:192
Definition reorderbuffer.c:161
Definition reorderbuffer.c:170
Definition reorderbuffer.c:131
Definition reorderbuffer.h:294
SharedInvalidationMessage * invalidations
Definition reorderbuffer.h:430
uint32 ninvalidations_distributed
Definition reorderbuffer.h:435
SharedInvalidationMessage * invalidations_distributed
Definition reorderbuffer.h:436
Definition reorderbuffer.c:179
Definition reorderbuffer.c:144
Definition reorderbuffer.c:138
Definition reorderbuffer.h:575
dlist_head txns_by_base_snapshot_lsn
Definition reorderbuffer.h:594
XLogRecPtr current_restart_decoding_lsn
Definition reorderbuffer.h:665
Definition slot.h:181
Definition resowner.c:113
Definition rewriteheap.c:193
Definition snapbuild_internal.h:27
Definition snapshot.h:139
Definition reorderbuffer.c:153
Definition tupdesc.h:149
Definition dirent.h:10
Definition ilist.h:178
Definition ilist.h:199
Definition ilist.h:138
Definition pairingheap.h:31
Definition win32_port.h:272
Definition regguts.h:331
Definition varatt.h:33
Definition varatt.h:58
bool TransactionIdDidCommit(TransactionId transactionId)
Definition transam.c:126
static bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition transam.h:263
static CompactAttribute * TupleDescCompactAttr(TupleDesc tupdesc, int i)
Definition tupdesc.h:195
Definition pg_list.h:46
static void SET_VARSIZE_COMPRESSED(void *PTR, Size len)
Definition varatt.h:446
static void SET_VARTAG_EXTERNAL(void *PTR, vartag_external tag)
Definition varatt.h:453
static Size VARATT_EXTERNAL_GET_EXTSIZE(varatt_external toast_pointer)
Definition varatt.h:507
static bool VARATT_EXTERNAL_IS_COMPRESSED(varatt_external toast_pointer)
Definition varatt.h:536
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition wait_event.h:67
void RollbackAndReleaseCurrentSubTransaction(void)
Definition xact.c:4850
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
Definition xlog_internal.h:102
◆ IsSpecConfirmOrAbort
| #define IsSpecConfirmOrAbort | ( | action | ) |
Value:
( \
(((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM) || \
((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT)) \
)
Definition at line 202 of file reorderbuffer.c.
◆ IsSpecInsert
| #define IsSpecInsert | ( | action | ) |
Value:
Definition at line 198 of file reorderbuffer.c.
◆ MAX_DISTR_INVAL_MSG_PER_TXN
| #define MAX_DISTR_INVAL_MSG_PER_TXN ((8 * 1024 * 1024) / sizeof(SharedInvalidationMessage)) |
Definition at line 126 of file reorderbuffer.c.
Typedef Documentation
◆ ReorderBufferDiskChange
◆ ReorderBufferIterTXNEntry
◆ ReorderBufferIterTXNState
◆ ReorderBufferToastEnt
◆ ReorderBufferTupleCidEnt
◆ ReorderBufferTupleCidKey
◆ ReorderBufferTXNByIdEnt
◆ RewriteMappingFile
◆ TXNEntryFile
| typedef struct TXNEntryFile TXNEntryFile |
Function Documentation
◆ ApplyLogicalMappingFile()
Definition at line 5357 of file reorderbuffer.c.
5358{
5362 LogicalRewriteMappingData map;
5363
5368 (errcode_for_file_access(),
5370
5371 while (true)
5372 {
5376 bool found;
5377
5378 /* be careful about padding */
5380
5381 /* read all mappings till the end of the file */
5384 pgstat_report_wait_end();
5385
5388 (errcode_for_file_access(),
5390 path)));
5392 break;
5395 (errcode_for_file_access(),
5397 path, readBytes,
5399
5402 &key.tid);
5403
5404
5407
5408 /* no existing mapping, no need to update */
5410 continue;
5411
5414 &key.tid);
5415
5418
5419 if (found)
5420 {
5421 /*
5422 * Make sure the existing mapping makes sense. We sometime update
5423 * old records that did not yet have a cmax (e.g. pg_class' own
5424 * entry while rewriting it) during rewrites, so allow that.
5425 */
5428 }
5429 else
5430 {
5431 /* update mapping */
5435 }
5436 }
5437
5440 (errcode_for_file_access(),
5442}
References Assert, CloseTransientFile(), ereport, errcode_for_file_access(), errmsg, ERROR, fb(), fd(), HASH_ENTER, HASH_FIND, hash_search(), InvalidCommandId, ItemPointerCopy(), MAXPGPATH, LogicalRewriteMappingData::new_locator, LogicalRewriteMappingData::new_tid, LogicalRewriteMappingData::old_locator, LogicalRewriteMappingData::old_tid, OpenTransientFile(), PG_BINARY, PG_LOGICAL_MAPPINGS_DIR, pgstat_report_wait_end(), pgstat_report_wait_start(), read, sprintf, and tuplecid_data.
Referenced by UpdateLogicalMappings().
◆ AssertChangeLsnOrder()
|
static |
Definition at line 1014 of file reorderbuffer.c.
1015{
1016#ifdef USE_ASSERT_CHECKING
1017 dlist_iter iter;
1019
1021 {
1023
1025
1029
1032
1034
1036 }
1037#endif
1038}
References Assert, ReorderBufferTXN::changes, dlist_iter::cur, dlist_container, dlist_foreach, ReorderBufferTXN::end_lsn, fb(), ReorderBufferTXN::first_lsn, and XLogRecPtrIsValid.
Referenced by ReorderBufferIterTXNInit().
◆ AssertTXNLsnOrder()
|
static |
Definition at line 943 of file reorderbuffer.c.
944{
945#ifdef USE_ASSERT_CHECKING
947 dlist_iter iter;
950
951 /*
952 * Skip the verification if we don't reach the LSN at which we start
953 * decoding the contents of transactions yet because until we reach the
954 * LSN, we could have transactions that don't have the association between
955 * the top-level transaction and subtransaction yet and consequently have
956 * the same LSN. We don't guarantee this association until we try to
957 * decode the actual contents of transaction. The ordering of the records
958 * prior to the start_decoding_at LSN should have been checked before the
959 * restart.
960 */
962 return;
963
965 {
967 iter.cur);
968
969 /* start LSN must be set */
971
972 /* If there is an end LSN, it must be higher than start LSN */
975
976 /* Current initial LSN must be strictly higher than previous */
979
980 /* known-as-subtxn txns must not be listed */
982
984 }
985
987 {
989 base_snapshot_node,
990 iter.cur);
991
992 /* base snapshot (and its LSN) must be set */
995
996 /* current LSN must be strictly higher than previous */
999
1000 /* known-as-subtxn txns must not be listed */
1002
1004 }
1005#endif
1006}
References Assert, dlist_iter::cur, dlist_container, dlist_foreach, XLogReaderState::EndRecPtr, fb(), InvalidXLogRecPtr, rbtxn_is_known_subxact, LogicalDecodingContext::reader, SnapBuildXactNeedsSkip(), LogicalDecodingContext::snapshot_builder, and XLogRecPtrIsValid.
Referenced by ReorderBufferAssignChild(), ReorderBufferGetOldestTXN(), ReorderBufferGetOldestXmin(), ReorderBufferSetBaseSnapshot(), and ReorderBufferTXNByXid().
◆ file_sort_by_lsn()
Definition at line 5459 of file reorderbuffer.c.
5460{
5463
5465}
References a, b, fb(), lfirst, and pg_cmp_u64().
Referenced by UpdateLogicalMappings().
◆ ReorderBufferAbort()
| void ReorderBufferAbort | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn, | ||
| TimestampTz | abort_time | ||
| ) |
Definition at line 3083 of file reorderbuffer.c.
3085{
3086 ReorderBufferTXN *txn;
3087
3089 false);
3090
3091 /* unknown, nothing to remove */
3093 return;
3094
3095 txn->abort_time = abort_time;
3096
3097 /* For streamed transactions notify the remote node about the abort. */
3099 {
3101
3102 /*
3103 * We might have decoded changes for this transaction that could load
3104 * the cache as per the current transaction's view (consider DDL's
3105 * happened in this transaction). We don't want the decoding of future
3106 * transactions to use those cache entries so execute only the inval
3107 * messages in this transaction.
3108 */
3111 txn->invalidations);
3112 }
3113
3114 /* cosmetic... */
3115 txn->final_lsn = lsn;
3116
3117 /* remove potential on-disk data, and deallocate */
3119}
References ReorderBufferTXN::abort_time, fb(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBufferImmediateInvalidation(), and ReorderBufferTXNByXid().
Referenced by DecodeAbort().
◆ ReorderBufferAbortOld()
| void ReorderBufferAbortOld | ( | ReorderBuffer * | rb, |
| TransactionId | oldestRunningXid | ||
| ) |
Definition at line 3129 of file reorderbuffer.c.
3130{
3132
3133 /*
3134 * Iterate through all (potential) toplevel TXNs and abort all that are
3135 * older than what possibly can be running. Once we've found the first
3136 * that is alive we stop, there might be some that acquired an xid earlier
3137 * but started writing later, but it's unlikely and they will be cleaned
3138 * up in a later call to this function.
3139 */
3141 {
3142 ReorderBufferTXN *txn;
3143
3145
3147 {
3149
3150 /* Notify the remote node about the crash/immediate restart. */
3153
3154 /* remove potential on-disk data, and deallocate this tx */
3156 }
3157 else
3158 return;
3159 }
3160}
References DEBUG2, dlist_container, dlist_foreach_modify, elog, fb(), InvalidXLogRecPtr, rbtxn_is_streamed, ReorderBufferCleanupTXN(), TransactionIdPrecedes(), and ReorderBufferTXN::xid.
Referenced by standby_decode().
◆ ReorderBufferAccumulateInvalidations()
|
static |
Definition at line 3500 of file reorderbuffer.c.
3504{
3506 {
3510 }
3511 else
3512 {
3513 /* Enlarge the array of inval messages */
3514 *invals_out =
3520 }
3521}
References fb(), memcpy(), palloc_array, and repalloc_array.
Referenced by ReorderBufferAddDistributedInvalidations(), and ReorderBufferAddInvalidations().
◆ ReorderBufferAddDistributedInvalidations()
| void ReorderBufferAddDistributedInvalidations | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn, | ||
| Size | nmsgs, | ||
| SharedInvalidationMessage * | msgs | ||
| ) |
Definition at line 3578 of file reorderbuffer.c.
3581{
3582 ReorderBufferTXN *txn;
3583 MemoryContext oldcontext;
3584
3586
3588
3589 /*
3590 * Collect all the invalidations under the top transaction, if available,
3591 * so that we can execute them all together. See comments
3592 * ReorderBufferAddInvalidations.
3593 */
3594 txn = rbtxn_get_toptxn(txn);
3595
3596 Assert(nmsgs > 0);
3597
3599 {
3600 /*
3601 * Check the transaction has enough space for storing distributed
3602 * invalidation messages.
3603 */
3605 {
3606 /*
3607 * Mark the invalidation message as overflowed and free up the
3608 * messages accumulated so far.
3609 */
3611
3613 {
3616 txn->ninvalidations_distributed = 0;
3617 }
3618 }
3619 else
3621 &txn->ninvalidations_distributed,
3622 msgs, nmsgs);
3623 }
3624
3625 /* Queue the invalidation messages into the transaction */
3627
3628 MemoryContextSwitchTo(oldcontext);
3629}
References Assert, fb(), ReorderBufferTXN::invalidations_distributed, MAX_DISTR_INVAL_MSG_PER_TXN, MemoryContextSwitchTo(), ReorderBufferTXN::ninvalidations_distributed, pfree(), RBTXN_DISTR_INVAL_OVERFLOWED, rbtxn_distr_inval_overflowed, rbtxn_get_toptxn, ReorderBufferAccumulateInvalidations(), ReorderBufferQueueInvalidations(), ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.
Referenced by SnapBuildDistributeSnapshotAndInval().
◆ ReorderBufferAddInvalidations()
| void ReorderBufferAddInvalidations | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn, | ||
| Size | nmsgs, | ||
| SharedInvalidationMessage * | msgs | ||
| ) |
Definition at line 3537 of file reorderbuffer.c.
3540{
3541 ReorderBufferTXN *txn;
3542 MemoryContext oldcontext;
3543
3545
3547
3548 /*
3549 * Collect all the invalidations under the top transaction, if available,
3550 * so that we can execute them all together. See comments atop this
3551 * function.
3552 */
3553 txn = rbtxn_get_toptxn(txn);
3554
3555 Assert(nmsgs > 0);
3556
3558 &txn->ninvalidations,
3559 msgs, nmsgs);
3560
3562
3563 MemoryContextSwitchTo(oldcontext);
3564}
References Assert, fb(), ReorderBufferTXN::invalidations, MemoryContextSwitchTo(), ReorderBufferTXN::ninvalidations, rbtxn_get_toptxn, ReorderBufferAccumulateInvalidations(), ReorderBufferQueueInvalidations(), and ReorderBufferTXNByXid().
Referenced by xact_decode().
◆ ReorderBufferAddNewCommandId()
| void ReorderBufferAddNewCommandId | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn, | ||
| CommandId | cid | ||
| ) |
Definition at line 3352 of file reorderbuffer.c.
3354{
3356
3359
3361}
References ReorderBufferChange::action, ReorderBufferChange::command_id, ReorderBufferChange::data, fb(), REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, ReorderBufferAllocChange(), and ReorderBufferQueueChange().
Referenced by SnapBuildProcessNewCid().
◆ ReorderBufferAddNewTupleCids()
| void ReorderBufferAddNewTupleCids | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn, | ||
| RelFileLocator | locator, | ||
| ItemPointerData | tid, | ||
| CommandId | cmin, | ||
| CommandId | cmax, | ||
| CommandId | combocid | ||
| ) |
Definition at line 3451 of file reorderbuffer.c.
3455{
3457 ReorderBufferTXN *txn;
3458
3460
3466 change->lsn = lsn;
3467 change->txn = txn;
3469
3471 txn->ntuplecids++;
3472}
References ReorderBufferChange::action, ReorderBufferChange::cmax, ReorderBufferChange::cmin, ReorderBufferChange::combocid, ReorderBufferChange::data, dlist_push_tail(), fb(), ReorderBufferChange::locator, ReorderBufferChange::lsn, ReorderBufferChange::node, ReorderBufferTXN::ntuplecids, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferAllocChange(), ReorderBufferTXNByXid(), ReorderBufferChange::tid, ReorderBufferChange::tuplecid, ReorderBufferTXN::tuplecids, and ReorderBufferChange::txn.
Referenced by SnapBuildProcessNewCid().
◆ ReorderBufferAddSnapshot()
| void ReorderBufferAddSnapshot | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn, | ||
| Snapshot | snap | ||
| ) |
Definition at line 3303 of file reorderbuffer.c.
3305{
3307
3310
3312}
References ReorderBufferChange::action, ReorderBufferChange::data, fb(), REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, ReorderBufferAllocChange(), ReorderBufferQueueChange(), and ReorderBufferChange::snapshot.
Referenced by SnapBuildDistributeSnapshotAndInval().
◆ ReorderBufferAllocate()
| ReorderBuffer * ReorderBufferAllocate | ( | void | ) |
Definition at line 325 of file reorderbuffer.c.
326{
327 ReorderBuffer *buffer;
330
332
333 /* allocate memory in own context, to have better accountability */
335 "ReorderBuffer",
336 ALLOCSET_DEFAULT_SIZES);
337
338 buffer =
340
342
344
346 "Change",
349
351 "TXN",
354
355 /*
356 * To minimize memory fragmentation caused by long-running transactions
357 * with changes spanning multiple memory blocks, we use a single
358 * fixed-size memory block for decoded tuple storage. The performance
359 * testing showed that the default memory block size maintains logical
360 * decoding performance without causing fragmentation due to concurrent
361 * transactions. One might think that we can use the max size as
362 * SLAB_LARGE_BLOCK_SIZE but the test also showed it doesn't help resolve
363 * the memory fragmentation.
364 */
366 "Tuples",
369 SLAB_DEFAULT_BLOCK_SIZE);
370
374
377
380
382 buffer->outbufsize = 0;
383 buffer->size = 0;
384
385 /* txn_heap is ordered by transaction size */
387
388 buffer->spillTxns = 0;
389 buffer->spillCount = 0;
390 buffer->spillBytes = 0;
391 buffer->streamTxns = 0;
392 buffer->streamCount = 0;
393 buffer->streamBytes = 0;
394 buffer->memExceededCount = 0;
395 buffer->totalTxns = 0;
396 buffer->totalBytes = 0;
397
399
403
404 /*
405 * Ensure there's no stale data from prior uses of this slot, in case some
406 * prior exit avoided calling ReorderBufferFree. Failure to do this can
407 * produce duplicated txns, and it's very cheap if there's nothing there.
408 */
410
411 return buffer;
412}
References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, ReorderBuffer::by_txn, ReorderBuffer::by_txn_last_txn, ReorderBuffer::by_txn_last_xid, ReorderBuffer::catchange_txns, ReorderBuffer::change_context, ReorderBuffer::context, ReorderBuffer::current_restart_decoding_lsn, CurrentMemoryContext, ReplicationSlot::data, dclist_init(), dlist_init(), fb(), GenerationContextCreate(), HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, InvalidTransactionId, InvalidXLogRecPtr, ReorderBuffer::memExceededCount, MemoryContextAlloc(), MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReorderBuffer::outbuf, ReorderBuffer::outbufsize, pairingheap_allocate(), ReorderBufferCleanupSerializedTXNs(), ReorderBufferTXNSizeCompare(), ReorderBuffer::size, SLAB_DEFAULT_BLOCK_SIZE, SlabContextCreate(), ReorderBuffer::spillBytes, ReorderBuffer::spillCount, ReorderBuffer::spillTxns, ReorderBuffer::streamBytes, ReorderBuffer::streamCount, ReorderBuffer::streamTxns, ReorderBuffer::toplevel_by_lsn, ReorderBuffer::totalBytes, ReorderBuffer::totalTxns, ReorderBuffer::tup_context, ReorderBuffer::txn_context, ReorderBuffer::txn_heap, and ReorderBuffer::txns_by_base_snapshot_lsn.
Referenced by StartupDecodingContext().
◆ ReorderBufferAllocChange()
| ReorderBufferChange * ReorderBufferAllocChange | ( | ReorderBuffer * | rb | ) |
Definition at line 508 of file reorderbuffer.c.
509{
510 ReorderBufferChange *change;
511
512 change = (ReorderBufferChange *)
514
516 return change;
517}
References fb(), and MemoryContextAlloc().
Referenced by DecodeDelete(), DecodeInsert(), DecodeMultiInsert(), DecodeSpecConfirm(), DecodeTruncate(), DecodeUpdate(), ReorderBufferAddNewCommandId(), ReorderBufferAddNewTupleCids(), ReorderBufferAddSnapshot(), ReorderBufferQueueInvalidations(), ReorderBufferQueueMessage(), and ReorderBufferRestoreChange().
◆ ReorderBufferAllocRelids()
| Oid * ReorderBufferAllocRelids | ( | ReorderBuffer * | rb, |
| int | nrelids | ||
| ) |
Definition at line 626 of file reorderbuffer.c.
References fb(), and MemoryContextAlloc().
Referenced by DecodeTruncate(), and ReorderBufferRestoreChange().
◆ ReorderBufferAllocTupleBuf()
| HeapTuple ReorderBufferAllocTupleBuf | ( | ReorderBuffer * | rb, |
| Size | tuple_len | ||
| ) |
Definition at line 593 of file reorderbuffer.c.
References fb(), HEAPTUPLESIZE, MemoryContextAlloc(), SizeofHeapTupleHeader, and HeapTupleData::t_data.
Referenced by DecodeDelete(), DecodeInsert(), DecodeMultiInsert(), DecodeUpdate(), and ReorderBufferRestoreChange().
◆ ReorderBufferAllocTXN()
|
static |
Definition at line 436 of file reorderbuffer.c.
437{
438 ReorderBufferTXN *txn;
439
440 txn = (ReorderBufferTXN *)
442
444
448
449 /* InvalidCommandId is not zero, so set it explicitly */
452
453 return txn;
454}
References ReorderBufferTXN::changes, ReorderBufferTXN::command_id, dlist_init(), fb(), InvalidCommandId, MemoryContextAlloc(), ReorderBufferTXN::output_plugin_private, ReorderBufferTXN::subtxns, and ReorderBufferTXN::tuplecids.
Referenced by ReorderBufferTXNByXid().
◆ ReorderBufferApplyChange()
|
inlinestatic |
Definition at line 2073 of file reorderbuffer.c.
2076{
2077 if (streaming)
2079 else
2081}
References fb().
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferApplyMessage()
|
inlinestatic |
Definition at line 2101 of file reorderbuffer.c.
2103{
2104 if (streaming)
2109 else
2114}
References ReorderBufferChange::data, fb(), ReorderBufferChange::lsn, ReorderBufferChange::message, ReorderBufferChange::message_size, ReorderBufferChange::msg, and ReorderBufferChange::prefix.
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferApplyTruncate()
|
inlinestatic |
Definition at line 2087 of file reorderbuffer.c.
2090{
2091 if (streaming)
2093 else
2095}
References fb().
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferAssignChild()
| void ReorderBufferAssignChild | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| TransactionId | subxid, | ||
| XLogRecPtr | lsn | ||
| ) |
Definition at line 1100 of file reorderbuffer.c.
1102{
1103 ReorderBufferTXN *txn;
1107
1110
1112 {
1114 {
1115 /* already associated, nothing to do */
1116 return;
1117 }
1118 else
1119 {
1120 /*
1121 * We already saw this transaction, but initially added it to the
1122 * list of top-level txns. Now that we know it's not top-level,
1123 * remove it from there.
1124 */
1126 }
1127 }
1128
1130 subtxn->toplevel_xid = xid;
1132
1133 /* set the reference to top-level transaction */
1134 subtxn->toptxn = txn;
1135
1136 /* add to subtransaction list */
1138 txn->nsubtxns++;
1139
1140 /* Possibly transfer the subtxn's snapshot to its top-level txn. */
1142
1143 /* Verify LSN-ordering invariant */
1145}
References Assert, AssertTXNLsnOrder(), dlist_delete(), dlist_push_tail(), fb(), ReorderBufferTXN::nsubtxns, rbtxn_is_known_subxact, RBTXN_IS_SUBXACT, ReorderBufferTransferSnapToParent(), ReorderBufferTXNByXid(), and ReorderBufferTXN::subtxns.
Referenced by LogicalDecodingProcessRecord(), and ReorderBufferCommitChild().
◆ ReorderBufferBuildTupleCidHash()
|
static |
Definition at line 1837 of file reorderbuffer.c.
1838{
1839 dlist_iter iter;
1841
1843 return;
1844
1848
1849 /*
1850 * create the hash with the exact number of to-be-stored tuplecids from
1851 * the start
1852 */
1853 txn->tuplecid_hash =
1856
1858 {
1861 bool found;
1862 ReorderBufferChange *change;
1863
1865
1867
1868 /* be careful about padding */
1870
1872
1874 &key.tid);
1875
1878 if (!found)
1879 {
1883 }
1884 else
1885 {
1886 /*
1887 * Maybe we already saw this tuple before in this transaction, but
1888 * if so it must have the same cmin.
1889 */
1891
1892 /*
1893 * cmax may be initially invalid, but once set it can only grow,
1894 * and never become invalid again.
1895 */
1900 }
1901 }
1902}
References ReorderBufferChange::action, Assert, ReorderBufferChange::cmax, ReorderBufferChange::cmin, ReorderBufferChange::combocid, dlist_iter::cur, ReorderBufferChange::data, dlist_container, dlist_foreach, dlist_is_empty(), fb(), HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASH_ENTER, hash_search(), InvalidCommandId, ItemPointerCopy(), HASHCTL::keysize, ReorderBufferChange::locator, ReorderBufferTXN::ntuplecids, rbtxn_has_catalog_changes, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferChange::tid, ReorderBufferChange::tuplecid, ReorderBufferTXN::tuplecid_hash, and ReorderBufferTXN::tuplecids.
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferCanStartStreaming()
|
inlinestatic |
Definition at line 4309 of file reorderbuffer.c.
4310{
4313
4314 /* We can't start streaming unless a consistent state is reached. */
4316 return false;
4317
4318 /*
4319 * We can't start streaming immediately even if the streaming is enabled
4320 * because we previously decoded this transaction and now just are
4321 * restarting.
4322 */
4325 return true;
4326
4327 return false;
4328}
References fb(), LogicalDecodingContext::reader, XLogReaderState::ReadRecPtr, ReorderBufferCanStream(), SNAPBUILD_CONSISTENT, SnapBuildCurrentState(), SnapBuildXactNeedsSkip(), and LogicalDecodingContext::snapshot_builder.
Referenced by ReorderBufferCheckMemoryLimit(), and ReorderBufferProcessPartialChange().
◆ ReorderBufferCanStream()
|
inlinestatic |
Definition at line 4300 of file reorderbuffer.c.
References fb(), and LogicalDecodingContext::streaming.
Referenced by ReorderBufferCanStartStreaming(), and ReorderBufferProcessPartialChange().
◆ ReorderBufferChangeMemoryUpdate()
|
static |
Definition at line 3380 of file reorderbuffer.c.
3384{
3385 ReorderBufferTXN *toptxn;
3386
3387 Assert(txn || change);
3388
3389 /*
3390 * Ignore tuple CID changes, because those are not evicted when reaching
3391 * memory limit. So we just don't count them, because it might easily
3392 * trigger a pointless attempt to spill.
3393 */
3395 return;
3396
3398 return;
3399
3401 txn = change->txn;
3403
3404 /*
3405 * Update the total size in top level as well. This is later used to
3406 * compute the decoding stats.
3407 */
3408 toptxn = rbtxn_get_toptxn(txn);
3409
3411 {
3413
3416
3417 /* Update the total size in the top transaction. */
3419
3420 /* Update the max-heap */
3424 }
3425 else
3426 {
3430
3431 /* Update the total size in the top transaction. */
3433
3434 /* Update the max-heap */
3438 }
3439
3441}
References ReorderBufferChange::action, Assert, fb(), pairingheap_add(), pairingheap_remove(), rbtxn_get_toptxn, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferTXN::size, ReorderBufferTXN::total_size, ReorderBufferChange::txn, and ReorderBufferTXN::txn_node.
Referenced by ReorderBufferCleanupTXN(), ReorderBufferFreeChange(), ReorderBufferQueueChange(), ReorderBufferRestoreChange(), ReorderBufferSerializeTXN(), ReorderBufferToastReplace(), and ReorderBufferTruncateTXN().
◆ ReorderBufferChangeSize()
|
static |
Definition at line 4452 of file reorderbuffer.c.
4453{
4455
4457 {
4458 /* fall through these, they're all similar enough */
4463 {
4465 newtup;
4468
4471
4473 {
4477 }
4478
4480 {
4484 }
4485
4486 break;
4487 }
4489 {
4491
4494
4495 break;
4496 }
4498 {
4501 break;
4502 }
4504 {
4506
4508
4512
4513 break;
4514 }
4516 {
4518
4519 break;
4520 }
4525 /* ReorderBufferChange contains everything important */
4526 break;
4527 }
4528
4530}
References ReorderBufferChange::action, ReorderBufferChange::data, fb(), ReorderBufferChange::inval, ReorderBufferChange::message_size, ReorderBufferChange::msg, ReorderBufferChange::newtuple, ReorderBufferChange::ninvalidations, ReorderBufferChange::nrelids, ReorderBufferChange::oldtuple, ReorderBufferChange::prefix, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferChange::snapshot, HeapTupleData::t_len, ReorderBufferChange::tp, ReorderBufferChange::truncate, and SnapshotData::xcnt.
Referenced by ReorderBufferCleanupTXN(), ReorderBufferFreeChange(), ReorderBufferQueueChange(), ReorderBufferRestoreChange(), ReorderBufferToastReplace(), and ReorderBufferTruncateTXN().
◆ ReorderBufferCheckAndTruncateAbortedTXN()
|
static |
Definition at line 1775 of file reorderbuffer.c.
1776{
1777 /* Quick return for regression tests */
1779 return false;
1780
1781 /*
1782 * Quick return if the transaction status is already known.
1783 */
1784
1786 return false;
1788 {
1789 /* Already-aborted transactions should not have any changes */
1791
1792 return true;
1793 }
1794
1795 /* Otherwise, check the transaction status using CLOG lookup */
1796
1798 return false;
1799
1801 {
1802 /*
1803 * Remember the transaction is committed so that we can skip CLOG
1804 * check next time, avoiding the pressure on CLOG lookup.
1805 */
1808 return false;
1809 }
1810
1811 /*
1812 * The transaction aborted. We discard both the changes collected so far
1813 * and the toast reconstruction data. The full cleanup will happen as part
1814 * of decoding ABORT record of this transaction.
1815 */
1818
1819 /* All changes should be discarded */
1821
1822 /*
1823 * Mark the transaction as aborted so we can ignore future changes of this
1824 * transaction.
1825 */
1828
1829 return true;
1830}
References Assert, DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE, debug_logical_replication_streaming, fb(), RBTXN_IS_ABORTED, rbtxn_is_aborted, RBTXN_IS_COMMITTED, rbtxn_is_committed, rbtxn_is_prepared, ReorderBufferToastReset(), ReorderBufferTruncateTXN(), ReorderBufferTXN::size, TransactionIdDidCommit(), TransactionIdIsInProgress(), ReorderBufferTXN::txn_flags, unlikely, and ReorderBufferTXN::xid.
Referenced by ReorderBufferCheckMemoryLimit().
◆ ReorderBufferCheckMemoryLimit()
|
static |
Definition at line 3889 of file reorderbuffer.c.
3890{
3891 ReorderBufferTXN *txn;
3893
3895 {
3896 /*
3897 * Update the statistics as the memory usage has reached the limit. We
3898 * report the statistics update later in this function since we can
3899 * update the slot statistics altogether while streaming or
3900 * serializing transactions in most cases.
3901 */
3902 rb->memExceededCount += 1;
3903 }
3905 {
3906 /*
3907 * Bail out if debug_logical_replication_streaming is buffered and we
3908 * haven't exceeded the memory limit.
3909 */
3910 return;
3911 }
3912
3913 /*
3914 * If debug_logical_replication_streaming is immediate, loop until there's
3915 * no change. Otherwise, loop until we reach under the memory limit. One
3916 * might think that just by evicting the largest (sub)transaction we will
3917 * come under the memory limit based on assumption that the selected
3918 * transaction is at least as large as the most recent change (which
3919 * caused us to go over the memory limit). However, that is not true
3920 * because a user can reduce the logical_decoding_work_mem to a smaller
3921 * value before the most recent change.
3922 */
3925 rb->size > 0))
3926 {
3927 /*
3928 * Pick the largest non-aborted transaction and evict it from memory
3929 * by streaming, if possible. Otherwise, spill to disk.
3930 */
3933 {
3934 /* we know there has to be one, because the size is not zero */
3938
3939 /* skip the transaction if aborted */
3941 continue;
3942
3944 }
3945 else
3946 {
3947 /*
3948 * Pick the largest transaction (or subtransaction) and evict it
3949 * from memory by serializing it to disk.
3950 */
3952
3953 /* we know there has to be one, because the size is not zero */
3954 Assert(txn);
3957
3958 /* skip the transaction if aborted */
3960 continue;
3961
3963 }
3964
3965 /*
3966 * After eviction, the transaction should have no entries in memory,
3967 * and should use 0 bytes for changes.
3968 */
3971
3972 /*
3973 * We've reported the memExceededCount update while streaming or
3974 * serializing the transaction.
3975 */
3977 }
3978
3981
3982 /* We must be under the memory limit now. */
3984}
References Assert, DEBUG_LOGICAL_REP_STREAMING_BUFFERED, DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE, debug_logical_replication_streaming, fb(), logical_decoding_work_mem, ReorderBufferTXN::nentries_mem, rbtxn_is_toptxn, ReorderBufferCanStartStreaming(), ReorderBufferCheckAndTruncateAbortedTXN(), ReorderBufferLargestStreamableTopTXN(), ReorderBufferLargestTXN(), ReorderBufferSerializeTXN(), ReorderBufferStreamTXN(), ReorderBufferTXN::size, ReorderBufferTXN::total_size, and UpdateDecodingStats().
Referenced by ReorderBufferQueueChange().
◆ ReorderBufferCleanupSerializedTXNs()
Definition at line 4877 of file reorderbuffer.c.
4878{
4883
4885
4886 /* we're only handling directories here, skip if it's not ours */
4888 return;
4889
4892 {
4893 /* only look at names that can be ours */
4895 {
4898 spill_de->d_name);
4899
4902 (errcode_for_file_access(),
4904 path, PG_REPLSLOT_DIR, slotname)));
4905 }
4906 }
4908}
References AllocateDir(), ereport, errcode_for_file_access(), errmsg, ERROR, fb(), FreeDir(), INFO, lstat, MAXPGPATH, PG_REPLSLOT_DIR, ReadDirExtended(), S_ISDIR, snprintf, and sprintf.
Referenced by ReorderBufferAllocate(), ReorderBufferFree(), and StartupReorderBuffer().
◆ ReorderBufferCleanupTXN()
|
static |
Definition at line 1536 of file reorderbuffer.c.
1537{
1538 bool found;
1539 dlist_mutable_iter iter;
1541
1542 /* cleanup subtransactions & their changes */
1544 {
1546
1548
1549 /*
1550 * Subtransactions are always associated to the toplevel TXN, even if
1551 * they originally were happening inside another subtxn, so we won't
1552 * ever recurse more than one level deep here.
1553 */
1556
1558 }
1559
1560 /* cleanup changes in the txn */
1562 {
1563 ReorderBufferChange *change;
1564
1566
1567 /* Check we're not mixing changes from different transactions. */
1569
1570 /*
1571 * Instead of updating the memory counter for individual changes, we
1572 * sum up the size of memory to free so we can update the memory
1573 * counter all together below. This saves costs of maintaining the
1574 * max-heap.
1575 */
1577
1579 }
1580
1581 /* Update the memory counter */
1583
1584 /*
1585 * Cleanup the tuplecids we stored for decoding catalog snapshot access.
1586 * They are always stored in the toplevel transaction.
1587 */
1589 {
1590 ReorderBufferChange *change;
1591
1593
1594 /* Check we're not mixing changes from different transactions. */
1597
1599 }
1600
1601 /*
1602 * Cleanup the base snapshot, if set.
1603 */
1605 {
1608 }
1609
1610 /*
1611 * Cleanup the snapshot for the last streamed run.
1612 */
1614 {
1617 }
1618
1619 /*
1620 * Remove TXN from its containing lists.
1621 *
1622 * Note: if txn is known as subxact, we are deleting the TXN from its
1623 * parent's list of known subxacts; this leaves the parent's nsubxacts
1624 * count too high, but we don't care. Otherwise, we are deleting the TXN
1625 * from the LSN-ordered list of toplevel TXNs. We remove the TXN from the
1626 * list of catalog modifying transactions as well.
1627 */
1631
1632 /* now remove reference from buffer */
1634 Assert(found);
1635
1636 /* remove entries spilled to disk */
1639
1640 /* deallocate */
1642}
References ReorderBufferChange::action, Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::base_snapshot_node, ReorderBufferTXN::catchange_node, ReorderBufferTXN::changes, dlist_mutable_iter::cur, dclist_delete_from(), dlist_container, dlist_delete(), dlist_foreach_modify, fb(), HASH_REMOVE, hash_search(), ReorderBufferTXN::node, rbtxn_has_catalog_changes, rbtxn_is_known_subxact, rbtxn_is_serialized, rbtxn_is_streamed, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferCleanupTXN(), ReorderBufferFreeChange(), ReorderBufferFreeSnap(), ReorderBufferFreeTXN(), ReorderBufferRestoreCleanup(), SnapBuildSnapDecRefcount(), ReorderBufferTXN::snapshot_now, ReorderBufferTXN::subtxns, ReorderBufferTXN::tuplecids, ReorderBufferChange::txn, and ReorderBufferTXN::xid.
Referenced by ReorderBufferAbort(), ReorderBufferAbortOld(), ReorderBufferCleanupTXN(), ReorderBufferFinishPrepared(), ReorderBufferForget(), ReorderBufferProcessTXN(), ReorderBufferReplay(), and ReorderBufferStreamCommit().
◆ ReorderBufferCommit()
| void ReorderBufferCommit | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | commit_lsn, | ||
| XLogRecPtr | end_lsn, | ||
| TimestampTz | commit_time, | ||
| ReplOriginId | origin_id, | ||
| XLogRecPtr | origin_lsn | ||
| ) |
Definition at line 2880 of file reorderbuffer.c.
2884{
2885 ReorderBufferTXN *txn;
2886
2888 false);
2889
2890 /* unknown transaction, nothing to replay */
2892 return;
2893
2895 origin_id, origin_lsn);
2896}
References fb(), InvalidXLogRecPtr, ReorderBufferReplay(), and ReorderBufferTXNByXid().
Referenced by DecodeCommit().
◆ ReorderBufferCommitChild()
| void ReorderBufferCommitChild | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| TransactionId | subxid, | ||
| XLogRecPtr | commit_lsn, | ||
| XLogRecPtr | end_lsn | ||
| ) |
Definition at line 1220 of file reorderbuffer.c.
1223{
1225
1228
1229 /*
1230 * No need to do anything if that subtxn didn't contain any changes
1231 */
1233 return;
1234
1235 subtxn->final_lsn = commit_lsn;
1236 subtxn->end_lsn = end_lsn;
1237
1238 /*
1239 * Assign this subxact as a child of the toplevel xact (no-op if already
1240 * done.)
1241 */
1243}
References fb(), InvalidXLogRecPtr, ReorderBufferAssignChild(), and ReorderBufferTXNByXid().
Referenced by DecodeCommit(), and DecodePrepare().
◆ ReorderBufferCopySnap()
|
static |
Definition at line 1910 of file reorderbuffer.c.
1912{
1914 dlist_iter iter;
1916 Size size;
1917
1921
1924
1927 snap->regd_count = 0;
1929
1931
1932 /*
1933 * snap->subxip contains all txids that belong to our transaction which we
1934 * need to check via cmin/cmax. That's why we store the toplevel
1935 * transaction in there as well.
1936 */
1939
1940 /*
1941 * txn->nsubtxns isn't decreased when subtransactions abort, so count
1942 * manually. Since it's an upper boundary it is safe to use it for the
1943 * allocation above.
1944 */
1945 snap->subxcnt = 1;
1946
1948 {
1950
1953 snap->subxcnt++;
1954 }
1955
1956 /* sort so we can bsearch() later */
1958
1959 /* store the specified current CommandId */
1961
1963}
References dlist_iter::cur, dlist_container, dlist_foreach, fb(), i, memcpy(), MemoryContextAllocZero(), ReorderBufferTXN::nsubtxns, qsort, ReorderBufferTXN::subtxns, ReorderBufferTXN::xid, and xidComparator().
Referenced by ReorderBufferProcessTXN(), ReorderBufferSaveTXNSnapshot(), and ReorderBufferStreamTXN().
◆ ReorderBufferExecuteInvalidations()
|
static |
Definition at line 3636 of file reorderbuffer.c.
References i, and LocalExecuteInvalidationMessage().
Referenced by ReorderBufferFinishPrepared(), and ReorderBufferProcessTXN().
◆ ReorderBufferFinishPrepared()
| void ReorderBufferFinishPrepared | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | commit_lsn, | ||
| XLogRecPtr | end_lsn, | ||
| XLogRecPtr | two_phase_at, | ||
| TimestampTz | commit_time, | ||
| ReplOriginId | origin_id, | ||
| XLogRecPtr | origin_lsn, | ||
| char * | gid, | ||
| bool | is_commit | ||
| ) |
Definition at line 2997 of file reorderbuffer.c.
3002{
3003 ReorderBufferTXN *txn;
3004 XLogRecPtr prepare_end_lsn;
3005 TimestampTz prepare_time;
3006
3008
3009 /* unknown transaction, nothing to do */
3011 return;
3012
3013 /*
3014 * By this time the txn has the prepare record information, remember it to
3015 * be later used for rollback.
3016 */
3017 prepare_end_lsn = txn->end_lsn;
3018 prepare_time = txn->prepare_time;
3019
3020 /* add the gid in the txn */
3022
3023 /*
3024 * It is possible that this transaction is not decoded at prepare time
3025 * either because by that time we didn't have a consistent snapshot, or
3026 * two_phase was not enabled, or it was decoded earlier but we have
3027 * restarted. We only need to send the prepare if it was not decoded
3028 * earlier. We don't need to decode the xact for aborts if it is not done
3029 * already.
3030 */
3032 {
3033 /*
3034 * txn must have been marked as a prepared transaction and skipped but
3035 * not sent a prepare. Also, the prepare info must have been updated
3036 * in txn even if we skip prepare.
3037 */
3041
3042 /*
3043 * By this time the txn has the prepare record information and it is
3044 * important to use that so that downstream gets the accurate
3045 * information. If instead, we have passed commit information here
3046 * then downstream can behave as it has already replayed commit
3047 * prepared after the restart.
3048 */
3051 }
3052
3053 txn->final_lsn = commit_lsn;
3054 txn->end_lsn = end_lsn;
3055 txn->commit_time = commit_time;
3056 txn->origin_id = origin_id;
3057 txn->origin_lsn = origin_lsn;
3058
3061 else
3063
3064 /* cleanup: make sure there's no cache pollution */
3066 txn->invalidations);
3068}
References Assert, ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, fb(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::gid, ReorderBufferTXN::invalidations, ReorderBufferTXN::ninvalidations, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBufferTXN::prepare_time, pstrdup(), RBTXN_IS_PREPARED, RBTXN_PREPARE_STATUS_MASK, RBTXN_SKIPPED_PREPARE, ReorderBufferCleanupTXN(), ReorderBufferExecuteInvalidations(), ReorderBufferReplay(), ReorderBufferTXNByXid(), ReorderBufferTXN::txn_flags, and XLogRecPtrIsValid.
Referenced by DecodeAbort(), and DecodeCommit().
◆ ReorderBufferForget()
| void ReorderBufferForget | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn | ||
| ) |
Definition at line 3176 of file reorderbuffer.c.
3177{
3178 ReorderBufferTXN *txn;
3179
3181 false);
3182
3183 /* unknown, nothing to forget */
3185 return;
3186
3187 /* this transaction mustn't be streamed */
3189
3190 /* cosmetic... */
3191 txn->final_lsn = lsn;
3192
3193 /*
3194 * Process only cache invalidation messages in this transaction if there
3195 * are any. Even if we're not interested in the transaction's contents, it
3196 * could have manipulated the catalog and we need to update the caches
3197 * according to that.
3198 */
3201 txn->invalidations);
3202 else
3204
3205 /* remove potential on-disk data, and deallocate */
3207}
References Assert, ReorderBufferTXN::base_snapshot, fb(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBufferImmediateInvalidation(), and ReorderBufferTXNByXid().
Referenced by DecodeCommit().
◆ ReorderBufferFree()
| void ReorderBufferFree | ( | ReorderBuffer * | rb | ) |
Definition at line 418 of file reorderbuffer.c.
419{
421
422 /*
423 * We free separately allocated data by entirely scrapping reorderbuffer's
424 * memory context.
425 */
426 MemoryContextDelete(context);
427
428 /* Free disk space used by unconsumed reorder buffers */
430}
References ReplicationSlot::data, fb(), MemoryContextDelete(), MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, and ReorderBufferCleanupSerializedTXNs().
Referenced by FreeDecodingContext().
◆ ReorderBufferFreeChange()
| void ReorderBufferFreeChange | ( | ReorderBuffer * | rb, |
| ReorderBufferChange * | change, | ||
| bool | upd_mem | ||
| ) |
Definition at line 523 of file reorderbuffer.c.
525{
526 /* update memory accounting info */
529 ReorderBufferChangeSize(change));
530
531 /* free contained data */
533 {
539 {
542 }
543
545 {
548 }
549 break;
557 break;
562 break;
565 {
568 }
569 break;
570 /* no data in addition to the struct itself */
573 {
576 }
577 break;
582 break;
583 }
584
585 pfree(change);
586}
References ReorderBufferChange::action, ReorderBufferChange::data, fb(), ReorderBufferChange::inval, ReorderBufferChange::invalidations, ReorderBufferChange::message, ReorderBufferChange::msg, ReorderBufferChange::newtuple, ReorderBufferChange::oldtuple, pfree(), ReorderBufferChange::prefix, ReorderBufferChange::relids, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferFreeRelids(), ReorderBufferFreeSnap(), ReorderBufferFreeTupleBuf(), ReorderBufferChange::snapshot, ReorderBufferChange::tp, and ReorderBufferChange::truncate.
Referenced by ReorderBufferCleanupTXN(), ReorderBufferIterTXNFinish(), ReorderBufferIterTXNNext(), ReorderBufferProcessTXN(), ReorderBufferQueueChange(), ReorderBufferRestoreChanges(), ReorderBufferSerializeTXN(), ReorderBufferToastReset(), and ReorderBufferTruncateTXN().
◆ ReorderBufferFreeRelids()
| void ReorderBufferFreeRelids | ( | ReorderBuffer * | rb, |
| Oid * | relids | ||
| ) |
Definition at line 642 of file reorderbuffer.c.
References pfree().
Referenced by ReorderBufferFreeChange().
◆ ReorderBufferFreeSnap()
|
static |
Definition at line 1969 of file reorderbuffer.c.
References fb(), pfree(), and SnapBuildSnapDecRefcount().
Referenced by ReorderBufferCleanupTXN(), ReorderBufferFreeChange(), ReorderBufferProcessTXN(), and ReorderBufferStreamTXN().
◆ ReorderBufferFreeTupleBuf()
Definition at line 611 of file reorderbuffer.c.
References pfree().
Referenced by ReorderBufferFreeChange().
◆ ReorderBufferFreeTXN()
|
static |
Definition at line 460 of file reorderbuffer.c.
461{
462 /* clean the lookup cache if we were cached (quite likely) */
464 {
467 }
468
469 /* free data that's contained */
470
472 {
475 }
476
478 {
481 }
482
484 {
487 }
488
490 {
493 }
494
495 /* Reset the toast hash */
497
498 /* All changes must be deallocated */
500
501 pfree(txn);
502}
References Assert, fb(), ReorderBufferTXN::gid, hash_destroy(), ReorderBufferTXN::invalidations, ReorderBufferTXN::invalidations_distributed, InvalidTransactionId, pfree(), ReorderBufferToastReset(), ReorderBufferTXN::size, ReorderBufferTXN::tuplecid_hash, and ReorderBufferTXN::xid.
Referenced by ReorderBufferCleanupTXN().
◆ ReorderBufferGetCatalogChangesXacts()
| TransactionId * ReorderBufferGetCatalogChangesXacts | ( | ReorderBuffer * | rb | ) |
Definition at line 3684 of file reorderbuffer.c.
3685{
3686 dlist_iter iter;
3688 size_t xcnt = 0;
3689
3690 /* Quick return if the list is empty */
3693
3694 /* Initialize XID array */
3697 {
3699 catchange_node,
3700 iter.cur);
3701
3703
3704 xids[xcnt++] = txn->xid;
3705 }
3706
3708
3710 return xids;
3711}
References Assert, dlist_iter::cur, dclist_container, dclist_count(), dclist_foreach, fb(), palloc_array, qsort, rbtxn_has_catalog_changes, ReorderBufferTXN::xid, and xidComparator().
Referenced by SnapBuildSerialize().
◆ ReorderBufferGetInvalidations()
| uint32 ReorderBufferGetInvalidations | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| SharedInvalidationMessage ** | msgs | ||
| ) |
Definition at line 5624 of file reorderbuffer.c.
5626{
5627 ReorderBufferTXN *txn;
5628
5630 false);
5631
5633 return 0;
5634
5635 *msgs = txn->invalidations;
5636
5638}
References fb(), ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, and ReorderBufferTXNByXid().
Referenced by SnapBuildDistributeSnapshotAndInval().
◆ ReorderBufferGetOldestTXN()
| ReorderBufferTXN * ReorderBufferGetOldestTXN | ( | ReorderBuffer * | rb | ) |
Definition at line 1045 of file reorderbuffer.c.
References Assert, AssertTXNLsnOrder(), dlist_head_element, dlist_is_empty(), fb(), ReorderBufferTXN::first_lsn, rbtxn_is_known_subxact, and XLogRecPtrIsValid.
Referenced by SnapBuildProcessRunningXacts().
◆ ReorderBufferGetOldestXmin()
| TransactionId ReorderBufferGetOldestXmin | ( | ReorderBuffer * | rb | ) |
Definition at line 1073 of file reorderbuffer.c.
References AssertTXNLsnOrder(), ReorderBufferTXN::base_snapshot, dlist_head_element, dlist_is_empty(), fb(), InvalidTransactionId, and SnapshotData::xmin.
Referenced by SnapBuildProcessRunningXacts().
◆ ReorderBufferImmediateInvalidation()
| void ReorderBufferImmediateInvalidation | ( | ReorderBuffer * | rb, |
| uint32 | ninvalidations, | ||
| SharedInvalidationMessage * | invalidations | ||
| ) |
Definition at line 3249 of file reorderbuffer.c.
3251{
3255
3258
3259 /*
3260 * Force invalidations to happen outside of a valid transaction - that way
3261 * entries will just be marked as invalid without accessing the catalog.
3262 * That's advantageous because we don't need to setup the full state
3263 * necessary for catalog access.
3264 */
3266 AbortCurrentTransaction();
3267
3270
3272 {
3275 CurrentResourceOwner = cowner;
3276 }
3277}
References AbortCurrentTransaction(), BeginInternalSubTransaction(), CurrentMemoryContext, CurrentResourceOwner, fb(), i, IsTransactionOrTransactionBlock(), LocalExecuteInvalidationMessage(), MemoryContextSwitchTo(), and RollbackAndReleaseCurrentSubTransaction().
Referenced by ReorderBufferAbort(), ReorderBufferForget(), ReorderBufferInvalidate(), and xact_decode().
◆ ReorderBufferInvalidate()
| void ReorderBufferInvalidate | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn | ||
| ) |
Definition at line 3218 of file reorderbuffer.c.
3219{
3220 ReorderBufferTXN *txn;
3221
3223 false);
3224
3225 /* unknown, nothing to do */
3227 return;
3228
3229 /*
3230 * Process cache invalidation messages if there are any. Even if we're not
3231 * interested in the transaction's contents, it could have manipulated the
3232 * catalog and we need to update the caches according to that.
3233 */
3236 txn->invalidations);
3237 else
3239}
References Assert, ReorderBufferTXN::base_snapshot, fb(), ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, ReorderBufferImmediateInvalidation(), and ReorderBufferTXNByXid().
Referenced by DecodePrepare().
◆ ReorderBufferIterCompare()
Definition at line 1262 of file reorderbuffer.c.
1263{
1267
1269 return 1;
1271 return 0;
1272 return -1;
1273}
References a, arg, b, DatumGetInt32(), and fb().
Referenced by ReorderBufferIterTXNInit().
◆ ReorderBufferIterTXNFinish()
|
static |
Definition at line 1505 of file reorderbuffer.c.
1507{
1508 Size off;
1509
1511 {
1514 }
1515
1516 /* free memory we might have "leaked" in the last *Next call */
1518 {
1519 ReorderBufferChange *change;
1520
1525 }
1526
1529}
References Assert, binaryheap_free(), dlist_container, dlist_is_empty(), dlist_pop_head_node(), fb(), FileClose(), pfree(), and ReorderBufferFreeChange().
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferIterTXNInit()
|
static |
Definition at line 1285 of file reorderbuffer.c.
1287{
1288 Size nr_txns = 0;
1291 Size off;
1292
1294
1295 /* Check ordering of changes in the toplevel transaction. */
1296 AssertChangeLsnOrder(txn);
1297
1298 /*
1299 * Calculate the size of our heap: one element for every transaction that
1300 * contains changes. (Besides the transactions already in the reorder
1301 * buffer, we count the one we were directly passed.)
1302 */
1304 nr_txns++;
1305
1307 {
1309
1311
1312 /* Check ordering of changes in this subtransaction. */
1314
1316 nr_txns++;
1317 }
1318
1319 /* allocate iteration state */
1324
1325 state->nr_txns = nr_txns;
1327
1329 {
1330 state->entries[off].file.vfd = -1;
1331 state->entries[off].segno = 0;
1332 }
1333
1334 /* allocate heap */
1336 ReorderBufferIterCompare,
1337 state);
1338
1339 /* Now that the state fields are initialized, it is safe to return it. */
1341
1342 /*
1343 * Now insert items into the binary heap, in an unordered fashion. (We
1344 * will run a heap assembly step at the end; this is more efficient.)
1345 */
1346
1347 off = 0;
1348
1349 /* add toplevel transaction if it contains changes */
1351 {
1353
1355 {
1356 /* serialize remaining changes */
1359 &state->entries[off].segno);
1360 }
1361
1363 &txn->changes);
1364
1367 state->entries[off].txn = txn;
1368
1370 }
1371
1372 /* add subtransactions if they contain changes */
1374 {
1376
1378
1380 {
1382
1384 {
1385 /* serialize remaining changes */
1388 &state->entries[off].file,
1389 &state->entries[off].segno);
1390 }
1392 &cur_txn->changes);
1393
1397
1399 }
1400 }
1401
1402 /* assemble a valid binary heap */
1404}
References AssertChangeLsnOrder(), binaryheap_add_unordered(), binaryheap_allocate(), binaryheap_build(), ReorderBufferTXN::changes, dlist_container, dlist_foreach, dlist_head_element, dlist_init(), fb(), Int32GetDatum(), MemoryContextAllocZero(), ReorderBufferTXN::nentries, rbtxn_is_serialized, ReorderBufferIterCompare(), ReorderBufferRestoreChanges(), ReorderBufferSerializeTXN(), and ReorderBufferTXN::subtxns.
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferIterTXNNext()
|
static |
Definition at line 1413 of file reorderbuffer.c.
1414{
1415 ReorderBufferChange *change;
1416 ReorderBufferIterTXNEntry *entry;
1417 int32 off;
1418
1419 /* nothing there anymore */
1422
1424 entry = &state->entries[off];
1425
1426 /* free memory we might have "leaked" in the previous *Next call */
1428 {
1433 }
1434
1435 change = entry->change;
1436
1437 /*
1438 * update heap with information about which transaction has the next
1439 * relevant change in LSN order
1440 */
1441
1442 /* there are in-memory changes */
1444 {
1448
1449 /* txn stays the same */
1452
1454 return change;
1455 }
1456
1457 /* try to load changes from disk */
1459 {
1460 /*
1461 * Ugly: restoring changes will reuse *Change records, thus delete the
1462 * current one from the per-tx list and only free in the next call.
1463 */
1466
1467 /*
1468 * Update the total bytes processed by the txn for which we are
1469 * releasing the current set of changes and restoring the new set of
1470 * changes.
1471 */
1474 &state->entries[off].segno))
1475 {
1476 /* successfully restored changes from disk */
1480
1484
1486 /* txn stays the same */
1490
1491 return change;
1492 }
1493 }
1494
1495 /* ok, no changes there anymore, remove */
1497
1498 return change;
1499}
References Assert, binaryheap_empty, binaryheap_first(), binaryheap_remove_first(), binaryheap_replace_first(), ReorderBufferIterTXNEntry::change, ReorderBufferTXN::changes, DatumGetInt32(), DEBUG2, dlist_container, dlist_delete(), dlist_has_next(), dlist_head_element, dlist_is_empty(), dlist_next_node(), dlist_pop_head_node(), dlist_push_tail(), elog, fb(), ReorderBufferIterTXNEntry::file, Int32GetDatum(), ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, next, ReorderBufferChange::node, ReorderBufferFreeChange(), ReorderBufferRestoreChanges(), ReorderBufferTXN::size, and ReorderBufferIterTXNEntry::txn.
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferLargestStreamableTopTXN()
|
static |
Definition at line 3839 of file reorderbuffer.c.
3840{
3841 dlist_iter iter;
3844
3845 /* Find the largest top-level transaction having a base snapshot. */
3847 {
3848 ReorderBufferTXN *txn;
3849
3851
3852 /* must not be a subtxn */
3854 /* base_snapshot must be set */
3856
3857 /* Don't consider these kinds of transactions for eviction. */
3859 !rbtxn_has_streamable_change(txn) ||
3860 rbtxn_is_aborted(txn))
3861 continue;
3862
3863 /* Find the largest of the eviction candidates. */
3865 (txn->total_size > 0))
3866 {
3867 largest = txn;
3869 }
3870 }
3871
3873}
References Assert, ReorderBufferTXN::base_snapshot, dlist_iter::cur, dlist_container, dlist_foreach, fb(), rbtxn_has_partial_change, rbtxn_has_streamable_change, rbtxn_is_aborted, rbtxn_is_known_subxact, and ReorderBufferTXN::total_size.
Referenced by ReorderBufferCheckMemoryLimit().
◆ ReorderBufferLargestTXN()
|
static |
Definition at line 3798 of file reorderbuffer.c.
3799{
3801
3802 /* Get the largest transaction from the max-heap */
3805
3809
3811}
References Assert, fb(), pairingheap_container, and pairingheap_first().
Referenced by ReorderBufferCheckMemoryLimit().
◆ ReorderBufferMaybeMarkTXNStreamed()
|
static |
Definition at line 2139 of file reorderbuffer.c.
2140{
2141 /*
2142 * The top-level transaction, is marked as streamed always, even if it
2143 * does not contain any changes (that is, when all the changes are in
2144 * subtransactions).
2145 *
2146 * For subtransactions, we only mark them as streamed when there are
2147 * changes in them.
2148 *
2149 * We do it this way because of aborts - we don't want to send aborts for
2150 * XIDs the downstream is not aware of. And of course, it always knows
2151 * about the top-level xact (we send the XID in all messages), but we
2152 * never stream XIDs of empty subxacts.
2153 */
2156}
References ReorderBufferTXN::nentries_mem, RBTXN_IS_STREAMED, rbtxn_is_toptxn, and ReorderBufferTXN::txn_flags.
Referenced by ReorderBufferProcessTXN(), and ReorderBufferTruncateTXN().
◆ ReorderBufferPrepare()
| void ReorderBufferPrepare | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| char * | gid | ||
| ) |
Definition at line 2956 of file reorderbuffer.c.
2958{
2959 ReorderBufferTXN *txn;
2960
2962 false);
2963
2964 /* unknown transaction, nothing to replay */
2966 return;
2967
2968 /*
2969 * txn must have been marked as a prepared transaction and must have
2970 * neither been skipped nor sent a prepare. Also, the prepare info must
2971 * have been updated in it by now.
2972 */
2975
2977
2980
2981 /*
2982 * Send a prepare if not already done so. This might occur if we have
2983 * detected a concurrent abort while replaying the non-streaming
2984 * transaction.
2985 */
2987 {
2990 }
2991}
References Assert, ReorderBufferTXN::end_lsn, fb(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::gid, InvalidXLogRecPtr, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBufferTXN::prepare_time, pstrdup(), RBTXN_IS_PREPARED, RBTXN_PREPARE_STATUS_MASK, RBTXN_SENT_PREPARE, rbtxn_sent_prepare, ReorderBufferReplay(), ReorderBufferTXNByXid(), ReorderBufferTXN::txn_flags, and XLogRecPtrIsValid.
Referenced by DecodePrepare().
◆ ReorderBufferProcessPartialChange()
|
static |
Definition at line 742 of file reorderbuffer.c.
745{
746 ReorderBufferTXN *toptxn;
747
748 /*
749 * The partial changes need to be processed only while streaming
750 * in-progress transactions.
751 */
753 return;
754
755 /* Get the top transaction. */
756 toptxn = rbtxn_get_toptxn(txn);
757
758 /*
759 * Indicate a partial change for toast inserts. The change will be
760 * considered as complete once we get the insert or update on the main
761 * table and we are sure that the pending toast chunks are not required
762 * anymore.
763 *
764 * If we allow streaming when there are pending toast chunks then such
765 * chunks won't be released till the insert (multi_insert) is complete and
766 * we expect the txn to have streamed all changes after streaming. This
767 * restriction is mainly to ensure the correctness of streamed
768 * transactions and it doesn't seem worth uplifting such a restriction
769 * just to allow this case because anyway we will stream the transaction
770 * once such an insert is complete.
771 */
778
779 /*
780 * Indicate a partial change for speculative inserts. The change will be
781 * considered as complete once we get the speculative confirm or abort
782 * token.
783 */
789
790 /*
791 * Stream the transaction if it is serialized before and the changes are
792 * now complete in the top-level transaction.
793 *
794 * The reason for doing the streaming of such a transaction as soon as we
795 * get the complete change for it is that previously it would have reached
796 * the memory threshold and wouldn't get streamed because of incomplete
797 * changes. Delaying such transactions would increase apply lag for them.
798 */
800 !(rbtxn_has_partial_change(toptxn)) &&
801 rbtxn_is_serialized(txn) &&
802 rbtxn_has_streamable_change(toptxn))
804}
References ReorderBufferChange::action, ReorderBufferChange::clear_toast_afterwards, ReorderBufferChange::data, fb(), IsInsertOrUpdate, IsSpecConfirmOrAbort, IsSpecInsert, rbtxn_get_toptxn, RBTXN_HAS_PARTIAL_CHANGE, rbtxn_has_partial_change, rbtxn_has_streamable_change, rbtxn_is_serialized, ReorderBufferCanStartStreaming(), ReorderBufferCanStream(), ReorderBufferStreamTXN(), ReorderBufferChange::tp, and ReorderBufferTXN::txn_flags.
Referenced by ReorderBufferQueueChange().
◆ ReorderBufferProcessTXN()
|
static |
Definition at line 2204 of file reorderbuffer.c.
2209{
2218
2219 /* build data to be able to lookup the CommandIds of catalog tuples */
2221
2222 /* setup the initial snapshot */
2224
2225 /*
2226 * Decoding needs access to syscaches et al., which in turn use
2227 * heavyweight locks and such. Thus we need to have enough state around to
2228 * keep track of those. The easiest way is to simply use a transaction
2229 * internally. That also allows us to easily enforce that nothing writes
2230 * to the database by checking for xid assignments.
2231 *
2232 * When we're called via the SQL SRF there's already a transaction
2233 * started, so start an explicit subtransaction there.
2234 */
2236
2237 PG_TRY();
2238 {
2239 ReorderBufferChange *change;
2241 * changes */
2242
2245 else
2246 StartTransactionCommand();
2247
2248 /*
2249 * We only need to send begin/begin-prepare for non-streamed
2250 * transactions.
2251 */
2252 if (!streaming)
2253 {
2256 else
2258 }
2259
2262 {
2264 Oid reloid;
2265
2266 CHECK_FOR_INTERRUPTS();
2267
2268 /*
2269 * We can't call start stream callback before processing first
2270 * change.
2271 */
2273 {
2274 if (streaming)
2275 {
2279 }
2280 }
2281
2282 /*
2283 * Enforce correct ordering of changes, merged from multiple
2284 * subtransactions. The changes may have the same LSN due to
2285 * MULTI_INSERT xlog records.
2286 */
2288
2290
2291 /*
2292 * Set the current xid to detect concurrent aborts. This is
2293 * required for the cases when we decode the changes before the
2294 * COMMIT record is processed.
2295 */
2297 {
2300 }
2301
2303 {
2305
2306 /*
2307 * Confirmation for speculative insertion arrived. Simply
2308 * use as a normal record. It'll be cleaned up at the end
2309 * of INSERT processing.
2310 */
2314 change = specinsert;
2316
2317 /* intentionally fall through */
2318 pg_fallthrough;
2322 Assert(snapshot_now);
2323
2326
2327 /*
2328 * Mapped catalog tuple without data, emitted while
2329 * catalog table was in the process of being rewritten. We
2330 * can fail to look up the relfilenumber, because the
2331 * relmapper has no "historic" view, in contrast to the
2332 * normal catalog during decoding. Thus repeated rewrites
2333 * can cause a lookup failure. That's OK because we do not
2334 * decode catalog changes anyway. Normally such tuples
2335 * would be skipped over below, but we can't identify
2336 * whether the table should be logically logged without
2337 * mapping the relfilenumber to the oid.
2338 */
2346 MAIN_FORKNUM).str);
2347
2348 relation = RelationIdGetRelation(reloid);
2349
2352 reloid,
2354 MAIN_FORKNUM).str);
2355
2358
2359 /*
2360 * Ignore temporary heaps created during DDL unless the
2361 * plugin has asked for them.
2362 */
2365
2366 /*
2367 * For now ignore sequence changes entirely. Most of the
2368 * time they don't log changes using records we
2369 * understand, so it doesn't make sense to handle the few
2370 * cases we do.
2371 */
2374
2375 /* user-triggered change */
2377 {
2380 streaming);
2381
2382 /*
2383 * Only clear reassembled toast chunks if we're sure
2384 * they're not required anymore. The creator of the
2385 * tuple tells us.
2386 */
2389 }
2390 /* we're not interested in toast deletions */
2392 {
2393 /*
2394 * Need to reassemble the full toasted Datum in
2395 * memory, to ensure the chunks don't get reused till
2396 * we're done remove it from the list of this
2397 * transaction's changes. Otherwise it will get
2398 * freed/reused while restoring spooled data from
2399 * disk.
2400 */
2402
2405 change);
2406 }
2407
2408 change_done:
2409
2410 /*
2411 * If speculative insertion was confirmed, the record
2412 * isn't needed anymore.
2413 */
2415 {
2418 }
2419
2421 {
2422 RelationClose(relation);
2423 relation = NULL;
2424 }
2425 break;
2426
2428
2429 /*
2430 * Speculative insertions are dealt with by delaying the
2431 * processing of the insert until the confirmation record
2432 * arrives. For that we simply unlink the record from the
2433 * chain, so it does not get freed/reused while restoring
2434 * spooled data from disk.
2435 *
2436 * This is safe in the face of concurrent catalog changes
2437 * because the relevant relation can't be changed between
2438 * speculative insertion and confirmation due to
2439 * CheckTableNotInUse() and locking.
2440 */
2441
2442 /* Previous speculative insertion must be aborted */
2444
2445 /* and memorize the pending insertion */
2447 specinsert = change;
2448 break;
2449
2451
2452 /*
2453 * Abort for speculative insertion arrived. So cleanup the
2454 * specinsert tuple and toast hash.
2455 *
2456 * Note that we get the spec abort change for each toast
2457 * entry but we need to perform the cleanup only the first
2458 * time we get it for the main table.
2459 */
2461 {
2462 /*
2463 * We must clean the toast hash before processing a
2464 * completely new tuple to avoid confusion about the
2465 * previous tuple's toast chunks.
2466 */
2469
2470 /* We don't need this record anymore. */
2473 }
2474 break;
2475
2477 {
2481 Relation *relations;
2482
2485 {
2487 Relation rel;
2488
2489 rel = RelationIdGetRelation(relid);
2490
2493
2495 continue;
2496
2497 relations[nrelations++] = rel;
2498 }
2499
2500 /* Apply the truncate. */
2502 relations, change,
2503 streaming);
2504
2507
2508 break;
2509 }
2510
2513 break;
2514
2516 /* Execute the invalidation messages locally */
2519 break;
2520
2522 /* get rid of the old */
2524
2525 if (snapshot_now->copied)
2526 {
2528 snapshot_now =
2530 txn, command_id);
2531 }
2532
2533 /*
2534 * Restored from disk, need to be careful not to double
2535 * free. We could introduce refcounting for that, but for
2536 * now this seems infrequent enough not to care.
2537 */
2539 {
2540 snapshot_now =
2542 txn, command_id);
2543 }
2544 else
2545 {
2547 }
2548
2549 /* and continue with the new one */
2551 break;
2552
2555
2557 {
2559
2560 if (!snapshot_now->copied)
2561 {
2562 /* we don't use the global one anymore */
2564 txn, command_id);
2565 }
2566
2567 snapshot_now->curcid = command_id;
2568
2571 }
2572
2573 break;
2574
2577 break;
2578 }
2579
2580 /*
2581 * It is possible that the data is not sent to downstream for a
2582 * long time either because the output plugin filtered it or there
2583 * is a DDL that generates a lot of data that is not processed by
2584 * the plugin. So, in such cases, the downstream can timeout. To
2585 * avoid that we try to send a keepalive message if required.
2586 * Trying to send a keepalive message after every change has some
2587 * overhead, but testing showed there is no noticeable overhead if
2588 * we do it after every ~100 changes.
2589 */
2590#define CHANGES_THRESHOLD 100
2591
2593 {
2595 changes_count = 0;
2596 }
2597 }
2598
2599 /* speculative insertion record must be freed by now */
2601
2602 /* clean up the iterator */
2605
2606 /*
2607 * Update total transaction count and total bytes processed by the
2608 * transaction and its subtransactions. Ensure to not count the
2609 * streamed transaction multiple times.
2610 *
2611 * Note that the statistics computation has to be done after
2612 * ReorderBufferIterTXNFinish as it releases the serialized change
2613 * which we have already accounted in ReorderBufferIterTXNNext.
2614 */
2616 rb->totalTxns++;
2617
2619
2620 /*
2621 * Done with current changes, send the last message for this set of
2622 * changes depending upon streaming mode.
2623 */
2624 if (streaming)
2625 {
2627 {
2630 }
2631 }
2632 else
2633 {
2634 /*
2635 * Call either PREPARE (for two-phase transactions) or COMMIT (for
2636 * regular ones).
2637 */
2639 {
2643 }
2644 else
2646 }
2647
2648 /* this is just a sanity check against bad output plugin behaviour */
2651 GetCurrentTransactionId());
2652
2653 /*
2654 * Remember the command ID and snapshot for the next set of changes in
2655 * streaming mode.
2656 */
2657 if (streaming)
2659 else if (snapshot_now->copied)
2661
2662 /* cleanup */
2664
2665 /*
2666 * Aborting the current (sub-)transaction as a whole has the right
2667 * semantics. We want all locks acquired in here to be released, not
2668 * reassigned to the parent and we do not want any database access
2669 * have persistent effects.
2670 */
2671 AbortCurrentTransaction();
2672
2673 /* make sure there's no cache pollution */
2675 {
2677 InvalidateSystemCaches();
2678 }
2679 else
2680 {
2683 txn->invalidations_distributed);
2684 }
2685
2687 {
2690 CurrentResourceOwner = cowner;
2691 }
2692
2693 /*
2694 * We are here due to one of the four reasons: 1. Decoding an
2695 * in-progress txn. 2. Decoding a prepared txn. 3. Decoding of a
2696 * prepared txn that was (partially) streamed. 4. Decoding a committed
2697 * txn.
2698 *
2699 * For 1, we allow truncation of txn data by removing the changes
2700 * already streamed but still keeping other things like invalidations,
2701 * snapshot, and tuplecids. For 2 and 3, we indicate
2702 * ReorderBufferTruncateTXN to do more elaborate truncation of txn
2703 * data as the entire transaction has been decoded except for commit.
2704 * For 4, as the entire txn has been decoded, we can fully clean up
2705 * the TXN reorder buffer.
2706 */
2708 {
2709 if (streaming)
2711
2713 /* Reset the CheckXidAlive */
2715 }
2716 else
2718 }
2719 PG_CATCH();
2720 {
2723
2724 /* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */
2727
2729
2730 /*
2731 * Force cache invalidation to happen outside of a valid transaction
2732 * to prevent catalog access as we just caught an error.
2733 */
2734 AbortCurrentTransaction();
2735
2736 /* make sure there's no cache pollution */
2738 {
2740 InvalidateSystemCaches();
2741 }
2742 else
2743 {
2746 txn->invalidations_distributed);
2747 }
2748
2750 {
2753 CurrentResourceOwner = cowner;
2754 }
2755
2756 /* Free the specinsert change before freeing the ReorderBufferTXN */
2758 {
2761 }
2762
2763 /*
2764 * The error code ERRCODE_TRANSACTION_ROLLBACK indicates a concurrent
2765 * abort of the (sub)transaction we are streaming or preparing. We
2766 * need to do the cleanup and return gracefully on this error, see
2767 * SetupCheckXidLive.
2768 *
2769 * This error code can be thrown by one of the callbacks we call
2770 * during decoding so we need to ensure that we return gracefully only
2771 * when we are sending the data in streaming mode and the streaming is
2772 * not finished yet or when we are sending the data out on a PREPARE
2773 * during a two-phase commit.
2774 */
2777 {
2778 /* curtxn must be set for streaming or prepared transactions */
2780
2781 /* Cleanup the temporary error state. */
2782 FlushErrorState();
2785
2786 /* Remember the transaction is aborted. */
2789
2790 /* Mark the transaction is streamed if appropriate */
2793
2794 /* Reset the TXN so that it is allowed to stream remaining data. */
2796 command_id, prev_lsn);
2797 }
2798 else
2799 {
2802 PG_RE_THROW();
2803 }
2804 }
2805 PG_END_TRY();
2806}
References AbortCurrentTransaction(), ReorderBufferChange::action, Assert, BeginInternalSubTransaction(), CHANGES_THRESHOLD, CHECK_FOR_INTERRUPTS, CheckXidAlive, ReorderBufferChange::clear_toast_afterwards, ReorderBufferChange::command_id, SnapshotData::copied, CopyErrorData(), SnapshotData::curcid, CurrentMemoryContext, CurrentResourceOwner, ReorderBufferChange::data, data, dlist_delete(), elog, ERROR, fb(), FlushErrorState(), FreeErrorData(), GetCurrentTransactionId(), GetCurrentTransactionIdIfAny(), i, ReorderBufferChange::inval, InvalidateSystemCaches(), ReorderBufferChange::invalidations, ReorderBufferTXN::invalidations, ReorderBufferTXN::invalidations_distributed, InvalidCommandId, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, IsToastRelation(), IsTransactionOrTransactionBlock(), ReorderBufferChange::lsn, MAIN_FORKNUM, MemoryContextSwitchTo(), ReorderBufferChange::newtuple, ReorderBufferChange::ninvalidations, ReorderBufferTXN::ninvalidations, ReorderBufferTXN::ninvalidations_distributed, ReorderBufferChange::node, ReorderBufferChange::nrelids, ReorderBufferChange::oldtuple, ReorderBufferChange::origin_id, ReorderBufferTXN::origin_id, palloc0_array, PG_CATCH, PG_END_TRY, pg_fallthrough, PG_RE_THROW, PG_TRY, rbtxn_distr_inval_overflowed, RBTXN_IS_ABORTED, rbtxn_is_committed, rbtxn_is_prepared, rbtxn_is_streamed, RBTXN_SENT_PREPARE, rbtxn_sent_prepare, RelationData::rd_rel, RelationClose(), RelationIdGetRelation(), RelationIsLogicallyLogged, RelationIsValid, RelidByRelfilenumber(), ReorderBufferChange::relids, RelFileLocator::relNumber, relpathperm, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferApplyChange(), ReorderBufferApplyMessage(), ReorderBufferApplyTruncate(), ReorderBufferBuildTupleCidHash(), ReorderBufferCleanupTXN(), ReorderBufferCopySnap(), ReorderBufferExecuteInvalidations(), ReorderBufferFreeChange(), ReorderBufferFreeSnap(), ReorderBufferIterTXNFinish(), ReorderBufferIterTXNInit(), ReorderBufferIterTXNNext(), ReorderBufferMaybeMarkTXNStreamed(), ReorderBufferResetTXN(), ReorderBufferSaveTXNSnapshot(), ReorderBufferToastAppendChunk(), ReorderBufferToastReplace(), ReorderBufferToastReset(), ReorderBufferTruncateTXN(), ReorderBufferChange::rlocator, RollbackAndReleaseCurrentSubTransaction(), SetupCheckXidLive(), SetupHistoricSnapshot(), ReorderBufferChange::snapshot, RelFileLocator::spcOid, StartTransactionCommand(), TeardownHistoricSnapshot(), ReorderBufferTXN::total_size, ReorderBufferChange::tp, ReorderBufferChange::truncate, ReorderBufferTXN::tuplecid_hash, ReorderBufferChange::txn, ReorderBufferTXN::txn_flags, and XLogRecPtrIsValid.
Referenced by ReorderBufferReplay(), and ReorderBufferStreamTXN().
◆ ReorderBufferProcessXid()
| void ReorderBufferProcessXid | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn | ||
| ) |
Definition at line 3290 of file reorderbuffer.c.
3291{
3292 /* many records won't have an xid assigned, centralize check here */
3295}
References fb(), InvalidTransactionId, and ReorderBufferTXNByXid().
Referenced by heap2_decode(), heap_decode(), LogicalDecodingProcessRecord(), logicalmsg_decode(), standby_decode(), xact_decode(), xlog2_decode(), and xlog_decode().
◆ ReorderBufferQueueChange()
| void ReorderBufferQueueChange | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn, | ||
| ReorderBufferChange * | change, | ||
| bool | toast_insert | ||
| ) |
Definition at line 811 of file reorderbuffer.c.
813{
814 ReorderBufferTXN *txn;
815
817
818 /*
819 * If we have detected that the transaction is aborted while streaming the
820 * previous changes or by checking its CLOG, there is no point in
821 * collecting further changes for it.
822 */
824 {
825 /*
826 * We don't need to update memory accounting for this change as we
827 * have not added it to the queue yet.
828 */
830 return;
831 }
832
833 /*
834 * The changes that are sent downstream are considered streamable. We
835 * remember such transactions so that only those will later be considered
836 * for streaming.
837 */
844 {
846
848 }
849
850 change->lsn = lsn;
851 change->txn = txn;
852
855 txn->nentries++;
856 txn->nentries_mem++;
857
858 /* update memory accounting information */
860 ReorderBufferChangeSize(change));
861
862 /* process partial change */
864
865 /* check the memory limits and evict something if needed */
867}
References ReorderBufferChange::action, Assert, ReorderBufferTXN::changes, dlist_push_tail(), fb(), ReorderBufferChange::lsn, ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, ReorderBufferChange::node, rbtxn_get_toptxn, RBTXN_HAS_STREAMABLE_CHANGE, rbtxn_is_aborted, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferCheckMemoryLimit(), ReorderBufferFreeChange(), ReorderBufferProcessPartialChange(), ReorderBufferTXNByXid(), ReorderBufferChange::txn, ReorderBufferTXN::txn_flags, and XLogRecPtrIsValid.
Referenced by DecodeDelete(), DecodeInsert(), DecodeMultiInsert(), DecodeSpecConfirm(), DecodeTruncate(), DecodeUpdate(), ReorderBufferAddNewCommandId(), ReorderBufferAddSnapshot(), ReorderBufferQueueInvalidations(), and ReorderBufferQueueMessage().
◆ ReorderBufferQueueInvalidations()
|
static |
Definition at line 3478 of file reorderbuffer.c.
References ReorderBufferChange::action, ReorderBufferChange::data, fb(), ReorderBufferChange::inval, ReorderBufferChange::invalidations, memcpy(), ReorderBufferChange::ninvalidations, palloc_array, REORDER_BUFFER_CHANGE_INVALIDATION, ReorderBufferAllocChange(), and ReorderBufferQueueChange().
Referenced by ReorderBufferAddDistributedInvalidations(), and ReorderBufferAddInvalidations().
◆ ReorderBufferQueueMessage()
| void ReorderBufferQueueMessage | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| Snapshot | snap, | ||
| XLogRecPtr | lsn, | ||
| bool | transactional, | ||
| const char * | prefix, | ||
| Size | message_size, | ||
| const char * | message | ||
| ) |
Definition at line 874 of file reorderbuffer.c.
878{
879 if (transactional)
880 {
881 MemoryContext oldcontext;
882 ReorderBufferChange *change;
883
885
886 /*
887 * We don't expect snapshots for transactional changes - we'll use the
888 * snapshot derived later during apply (unless the change gets
889 * skipped).
890 */
892
894
901
903
904 MemoryContextSwitchTo(oldcontext);
905 }
906 else
907 {
910
911 /* Non-transactional changes require a valid snapshot. */
912 Assert(snapshot_now);
913
916
917 /* setup snapshot to allow catalog access */
919 PG_TRY();
920 {
922
924 }
925 PG_CATCH();
926 {
928 PG_RE_THROW();
929 }
930 PG_END_TRY();
931 }
932}
References ReorderBufferChange::action, Assert, ReorderBufferChange::data, fb(), InvalidTransactionId, memcpy(), MemoryContextSwitchTo(), ReorderBufferChange::message, ReorderBufferChange::message_size, ReorderBufferChange::msg, palloc(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, ReorderBufferChange::prefix, pstrdup(), REORDER_BUFFER_CHANGE_MESSAGE, ReorderBufferAllocChange(), ReorderBufferQueueChange(), ReorderBufferTXNByXid(), SetupHistoricSnapshot(), and TeardownHistoricSnapshot().
Referenced by logicalmsg_decode().
◆ ReorderBufferRememberPrepareInfo()
| bool ReorderBufferRememberPrepareInfo | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | prepare_lsn, | ||
| XLogRecPtr | end_lsn, | ||
| TimestampTz | prepare_time, | ||
| ReplOriginId | origin_id, | ||
| XLogRecPtr | origin_lsn | ||
| ) |
Definition at line 2903 of file reorderbuffer.c.
2907{
2908 ReorderBufferTXN *txn;
2909
2911
2912 /* unknown transaction, nothing to do */
2914 return false;
2915
2916 /*
2917 * Remember the prepare information to be later used by commit prepared in
2918 * case we skip doing prepare.
2919 */
2920 txn->final_lsn = prepare_lsn;
2921 txn->end_lsn = end_lsn;
2922 txn->prepare_time = prepare_time;
2923 txn->origin_id = origin_id;
2924 txn->origin_lsn = origin_lsn;
2925
2926 /* Mark this transaction as a prepared transaction */
2929
2930 return true;
2931}
References Assert, ReorderBufferTXN::end_lsn, fb(), ReorderBufferTXN::final_lsn, InvalidXLogRecPtr, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBufferTXN::prepare_time, RBTXN_IS_PREPARED, RBTXN_PREPARE_STATUS_MASK, ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.
Referenced by DecodePrepare().
◆ ReorderBufferReplay()
|
static |
Definition at line 2819 of file reorderbuffer.c.
2824{
2825 Snapshot snapshot_now;
2827
2828 txn->final_lsn = commit_lsn;
2829 txn->end_lsn = end_lsn;
2830 txn->commit_time = commit_time;
2831 txn->origin_id = origin_id;
2832 txn->origin_lsn = origin_lsn;
2833
2834 /*
2835 * If the transaction was (partially) streamed, we need to commit it in a
2836 * 'streamed' way. That is, we first stream the remaining part of the
2837 * transaction, and then invoke stream_commit message.
2838 *
2839 * Called after everything (origin ID, LSN, ...) is stored in the
2840 * transaction to avoid passing that information directly.
2841 */
2843 {
2845 return;
2846 }
2847
2848 /*
2849 * If this transaction has no snapshot, it didn't make any changes to the
2850 * database, so there's nothing to decode. Note that
2851 * ReorderBufferCommitChild will have transferred any snapshots from
2852 * subtransactions if there were any.
2853 */
2855 {
2857
2858 /*
2859 * Removing this txn before a commit might result in the computation
2860 * of an incorrect restart_lsn. See SnapBuildProcessRunningXacts.
2861 */
2864 return;
2865 }
2866
2867 snapshot_now = txn->base_snapshot;
2868
2869 /* Process and send the changes to output plugin. */
2871 command_id, false);
2872}
References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, fb(), ReorderBufferTXN::final_lsn, FirstCommandId, ReorderBufferTXN::ninvalidations, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, rbtxn_is_prepared, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBufferProcessTXN(), and ReorderBufferStreamCommit().
Referenced by ReorderBufferCommit(), ReorderBufferFinishPrepared(), and ReorderBufferPrepare().
◆ ReorderBufferResetTXN()
|
static |
Definition at line 2166 of file reorderbuffer.c.
2170{
2171 /* Discard the changes that we just streamed */
2173
2174 /* Free all resources allocated for toast reconstruction */
2176
2177 /*
2178 * For the streaming case, stop the stream and remember the command ID and
2179 * snapshot for the streaming run.
2180 */
2182 {
2185 }
2186
2187 /* All changes must be deallocated */
2189}
References Assert, fb(), rbtxn_is_prepared, rbtxn_is_streamed, ReorderBufferSaveTXNSnapshot(), ReorderBufferToastReset(), ReorderBufferTruncateTXN(), and ReorderBufferTXN::size.
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferRestoreChange()
|
static |
Definition at line 4680 of file reorderbuffer.c.
4682{
4683 ReorderBufferDiskChange *ondisk;
4684 ReorderBufferChange *change;
4685
4687
4689
4690 /* copy static part */
4692
4694
4695 /* restore individual stuff */
4697 {
4698 /* fall through these, they're all similar enough */
4704 {
4706
4709
4710 /* restore ->tuple */
4714
4715 /* reset t_data pointer into the new tuplebuf */
4718
4719 /* restore tuple data itself */
4722 }
4723
4725 {
4726 /* here, data might not be suitably aligned! */
4728
4731
4734
4735 /* restore ->tuple */
4739
4740 /* reset t_data pointer into the new tuplebuf */
4743
4744 /* restore tuple data itself */
4747 }
4748
4749 break;
4751 {
4752 Size prefix_size;
4753
4754 /* read prefix */
4758 prefix_size);
4761 data += prefix_size;
4762
4763 /* read the message */
4771
4772 break;
4773 }
4775 {
4778
4781
4782 /* read the message */
4784
4785 break;
4786 }
4788 {
4791 Size size;
4792
4794
4798
4800
4802
4808 break;
4809 }
4810 /* the base struct contains all the data, easy peasy */
4812 {
4813 Oid *relids;
4814
4818
4819 break;
4820 }
4825 break;
4826 }
4827
4829 txn->nentries_mem++;
4830
4831 /*
4832 * Update memory accounting for the restored change. We need to do this
4833 * although we don't check the memory limit when restoring the changes in
4834 * this branch (we only do that when initially queueing the changes after
4835 * decoding), because we will release the changes later, and that will
4836 * update the accounting too (subtracting the size from the counters). And
4837 * we don't want to underflow there.
4838 */
4840 ReorderBufferChangeSize(change));
4841}
References ReorderBufferChange::action, Assert, ReorderBufferDiskChange::change, ReorderBufferTXN::changes, ReorderBufferChange::data, data, dlist_push_tail(), fb(), HEAPTUPLESIZE, ReorderBufferChange::inval, ReorderBufferChange::invalidations, memcpy(), MemoryContextAlloc(), MemoryContextAllocZero(), ReorderBufferChange::message, ReorderBufferChange::message_size, ReorderBufferChange::msg, ReorderBufferTXN::nentries_mem, ReorderBufferChange::newtuple, ReorderBufferChange::ninvalidations, ReorderBufferChange::node, ReorderBufferChange::nrelids, ReorderBufferChange::oldtuple, ReorderBufferChange::prefix, ReorderBufferChange::relids, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferAllocChange(), ReorderBufferAllocRelids(), ReorderBufferAllocTupleBuf(), ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), SizeofHeapTupleHeader, ReorderBufferChange::snapshot, HeapTupleData::t_data, ReorderBufferChange::tp, ReorderBufferChange::truncate, and SnapshotData::xcnt.
Referenced by ReorderBufferRestoreChanges().
◆ ReorderBufferRestoreChanges()
|
static |
Definition at line 4537 of file reorderbuffer.c.
4539{
4544
4547
4548 /* free current entries, so we have memory for more */
4550 {
4553
4556 }
4557 txn->nentries_mem = 0;
4559
4561
4563 {
4565 ReorderBufferDiskChange *ondisk;
4566
4567 CHECK_FOR_INTERRUPTS();
4568
4570 {
4572
4573 /* first time in */
4574 if (*segno == 0)
4576
4578
4579 /*
4580 * No need to care about TLIs here, only used during a single run,
4581 * so each LSN only maps to a specific WAL record.
4582 */
4584 *segno);
4585
4587
4588 /* No harm in resetting the offset even in case of failure */
4589 file->curOffset = 0;
4590
4592 {
4593 *fd = -1;
4594 (*segno)++;
4595 continue;
4596 }
4599 (errcode_for_file_access(),
4601 path)));
4602 }
4603
4604 /*
4605 * Read the statically sized part of a change which has information
4606 * about the total size. If we couldn't read a record, we're at the
4607 * end of this file.
4608 */
4613
4614 /* eof */
4616 {
4618 *fd = -1;
4619 (*segno)++;
4620 continue;
4621 }
4624 (errcode_for_file_access(),
4628 (errcode_for_file_access(),
4630 readBytes,
4632
4634
4636
4640
4644 file->curOffset,
4646
4649 (errcode_for_file_access(),
4653 (errcode_for_file_access(),
4655 readBytes,
4657
4659
4660 /*
4661 * ok, read a full change from disk, now restore it into proper
4662 * in-memory format
4663 */
4665 restored++;
4666 }
4667
4669}
References Assert, ReorderBufferTXN::changes, CHECK_FOR_INTERRUPTS, cleanup(), TXNEntryFile::curOffset, dlist_container, dlist_delete(), dlist_foreach_modify, dlist_is_empty(), ereport, errcode_for_file_access(), errmsg, ERROR, fb(), fd(), FileClose(), FileRead(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::first_lsn, max_changes_in_memory, MAXPGPATH, MyReplicationSlot, ReorderBufferTXN::nentries_mem, PathNameOpenFile(), PG_BINARY, ReorderBufferFreeChange(), ReorderBufferRestoreChange(), ReorderBufferSerializedPath(), ReorderBufferSerializeReserve(), ReorderBufferDiskChange::size, TXNEntryFile::vfd, wal_segment_size, ReorderBufferTXN::xid, XLByteToSeg, and XLogRecPtrIsValid.
Referenced by ReorderBufferIterTXNInit(), and ReorderBufferIterTXNNext().
◆ ReorderBufferRestoreCleanup()
|
static |
Definition at line 4847 of file reorderbuffer.c.
4848{
4849 XLogSegNo first;
4851 XLogSegNo last;
4852
4855
4858
4859 /* iterate over all possible filenames, and delete them */
4861 {
4863
4867 (errcode_for_file_access(),
4869 }
4870}
References Assert, cur, ereport, errcode_for_file_access(), errmsg, ERROR, fb(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::first_lsn, MAXPGPATH, MyReplicationSlot, ReorderBufferSerializedPath(), wal_segment_size, ReorderBufferTXN::xid, XLByteToSeg, and XLogRecPtrIsValid.
Referenced by ReorderBufferCleanupTXN(), and ReorderBufferTruncateTXN().
◆ ReorderBufferSaveTXNSnapshot()
|
inlinestatic |
Definition at line 2121 of file reorderbuffer.c.
2123{
2124 txn->command_id = command_id;
2125
2126 /* Avoid copying if it's already copied. */
2128 txn->snapshot_now = snapshot_now;
2129 else
2131 txn, command_id);
2132}
References ReorderBufferTXN::command_id, SnapshotData::copied, fb(), ReorderBufferCopySnap(), and ReorderBufferTXN::snapshot_now.
Referenced by ReorderBufferProcessTXN(), and ReorderBufferResetTXN().
◆ ReorderBufferSerializeChange()
|
static |
Definition at line 4085 of file reorderbuffer.c.
4087{
4088 ReorderBufferDiskChange *ondisk;
4090
4092
4095
4097 {
4098 /* fall through these, they're all similar enough */
4103 {
4106 newtup;
4109
4112
4114 {
4118 }
4119
4121 {
4125 }
4126
4127 /* make sure we have enough space */
4129
4131 /* might have been reallocated above */
4133
4135 {
4138
4141 }
4142
4144 {
4147
4150 }
4151 break;
4152 }
4154 {
4157
4161
4163
4164 /* might have been reallocated above */
4166
4167 /* write the prefix including the size */
4171 prefix_size);
4172 data += prefix_size;
4173
4174 /* write the message including the size */
4180
4181 break;
4182 }
4184 {
4188
4190
4193
4194 /* might have been reallocated above */
4198
4199 break;
4200 }
4202 {
4205
4207
4211
4212 /* make sure we have enough space */
4215 /* might have been reallocated above */
4217
4220
4222 {
4226 }
4227
4229 {
4233 }
4234 break;
4235 }
4237 {
4238 Size size;
4240
4241 /* account for the OIDs of truncated relations */
4243 sz += size;
4244
4245 /* make sure we have enough space */
4247
4249 /* might have been reallocated above */
4251
4253 data += size;
4254
4255 break;
4256 }
4261 /* ReorderBufferChange contains everything important */
4262 break;
4263 }
4264
4265 ondisk->size = sz;
4266
4267 errno = 0;
4270 {
4272
4274
4275 /* if write didn't set errno, assume problem is no disk space */
4278 (errcode_for_file_access(),
4280 txn->xid)));
4281 }
4282 pgstat_report_wait_end();
4283
4284 /*
4285 * Keep the transaction's final_lsn up to date with each change we send to
4286 * disk, so that ReorderBufferRestoreCleanup works correctly. (We used to
4287 * only do this on commit and abort records, but that doesn't work if a
4288 * system crash leaves a transaction without its abort record).
4289 *
4290 * Make sure not to move it backwards.
4291 */
4294
4296}
References ReorderBufferChange::action, Assert, ReorderBufferDiskChange::change, CloseTransientFile(), ReorderBufferChange::data, data, ereport, errcode_for_file_access(), errmsg, ERROR, fb(), fd(), ReorderBufferTXN::final_lsn, ReorderBufferChange::inval, ReorderBufferChange::invalidations, ReorderBufferChange::lsn, memcpy(), ReorderBufferChange::message, ReorderBufferChange::message_size, ReorderBufferChange::msg, ReorderBufferChange::newtuple, ReorderBufferChange::ninvalidations, ReorderBufferChange::nrelids, ReorderBufferChange::oldtuple, pgstat_report_wait_end(), pgstat_report_wait_start(), ReorderBufferChange::prefix, ReorderBufferChange::relids, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferSerializeReserve(), ReorderBufferDiskChange::size, ReorderBufferChange::snapshot, HeapTupleData::t_len, ReorderBufferChange::tp, ReorderBufferChange::truncate, write, SnapshotData::xcnt, and ReorderBufferTXN::xid.
Referenced by ReorderBufferSerializeTXN().
◆ ReorderBufferSerializedPath()
|
static |
Definition at line 4916 of file reorderbuffer.c.
References ReplicationSlot::data, fb(), LSN_FORMAT_ARGS, MAXPGPATH, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, PG_REPLSLOT_DIR, snprintf, wal_segment_size, and XLogSegNoOffsetToRecPtr.
Referenced by ReorderBufferRestoreChanges(), ReorderBufferRestoreCleanup(), and ReorderBufferSerializeTXN().
◆ ReorderBufferSerializeReserve()
|
static |
Definition at line 3765 of file reorderbuffer.c.
3766{
3768 {
3771 }
3773 {
3776 }
3777}
References fb(), MemoryContextAlloc(), and repalloc().
Referenced by ReorderBufferRestoreChanges(), and ReorderBufferSerializeChange().
◆ ReorderBufferSerializeTXN()
|
static |
Definition at line 3990 of file reorderbuffer.c.
3991{
3996 Size spilled = 0;
3998
4001
4002 /* do the same to all child TXs */
4004 {
4006
4009 }
4010
4011 /* serialize changestream */
4013 {
4014 ReorderBufferChange *change;
4015
4017
4018 /*
4019 * store in segment in which it belongs by start lsn, don't split over
4020 * multiple segments tho
4021 */
4024 {
4026
4029
4031
4032 /*
4033 * No need to care about TLIs here, only used during a single run,
4034 * so each LSN only maps to a specific WAL record.
4035 */
4037 curOpenSegNo);
4038
4039 /* open segment, create it if necessary */
4042
4045 (errcode_for_file_access(),
4047 }
4048
4052
4053 spilled++;
4054 }
4055
4056 /* Update the memory counter */
4058
4059 /* update the statistics iff we have spilled anything */
4060 if (spilled)
4061 {
4062 rb->spillCount += 1;
4063 rb->spillBytes += size;
4064
4065 /* don't consider already serialized transactions */
4067
4068 /* update the decoding stats */
4070 }
4071
4074 txn->nentries_mem = 0;
4076
4079}
References Assert, ReorderBufferTXN::changes, CloseTransientFile(), DEBUG2, dlist_container, dlist_delete(), dlist_foreach, dlist_foreach_modify, dlist_is_empty(), elog, ereport, errcode_for_file_access(), errmsg, ERROR, fb(), fd(), ReorderBufferChange::lsn, MAXPGPATH, MyReplicationSlot, ReorderBufferTXN::nentries_mem, ReorderBufferChange::node, OpenTransientFile(), PG_BINARY, RBTXN_IS_SERIALIZED, rbtxn_is_serialized, rbtxn_is_serialized_clear, ReorderBufferChangeMemoryUpdate(), ReorderBufferFreeChange(), ReorderBufferSerializeChange(), ReorderBufferSerializedPath(), ReorderBufferSerializeTXN(), ReorderBufferTXN::size, ReorderBufferTXN::subtxns, ReorderBufferTXN::txn_flags, UpdateDecodingStats(), wal_segment_size, ReorderBufferTXN::xid, XLByteInSeg, and XLByteToSeg.
Referenced by ReorderBufferCheckMemoryLimit(), ReorderBufferIterTXNInit(), and ReorderBufferSerializeTXN().
◆ ReorderBufferSetBaseSnapshot()
| void ReorderBufferSetBaseSnapshot | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn, | ||
| Snapshot | snap | ||
| ) |
Definition at line 3321 of file reorderbuffer.c.
3323{
3324 ReorderBufferTXN *txn;
3326
3328
3329 /*
3330 * Fetch the transaction to operate on. If we know it's a subtransaction,
3331 * operate on its top-level transaction instead.
3332 */
3338
3340 txn->base_snapshot_lsn = lsn;
3342
3344}
References Assert, AssertTXNLsnOrder(), ReorderBufferTXN::base_snapshot, ReorderBufferTXN::base_snapshot_lsn, ReorderBufferTXN::base_snapshot_node, dlist_push_tail(), fb(), InvalidXLogRecPtr, rbtxn_is_known_subxact, ReorderBufferTXNByXid(), and ReorderBufferTXN::toplevel_xid.
Referenced by SnapBuildCommitTxn(), and SnapBuildProcessChange().
◆ ReorderBufferSetRestartPoint()
| void ReorderBufferSetRestartPoint | ( | ReorderBuffer * | rb, |
| XLogRecPtr | ptr | ||
| ) |
Definition at line 1088 of file reorderbuffer.c.
References fb().
Referenced by SnapBuildRestore(), and SnapBuildSerialize().
◆ ReorderBufferSkipPrepare()
| void ReorderBufferSkipPrepare | ( | ReorderBuffer * | rb, |
| TransactionId | xid | ||
| ) |
Definition at line 2935 of file reorderbuffer.c.
2936{
2937 ReorderBufferTXN *txn;
2938
2940
2941 /* unknown transaction, nothing to do */
2943 return;
2944
2945 /* txn must have been marked as a prepared transaction */
2948}
References Assert, fb(), InvalidXLogRecPtr, RBTXN_IS_PREPARED, RBTXN_PREPARE_STATUS_MASK, RBTXN_SKIPPED_PREPARE, ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.
Referenced by DecodePrepare().
◆ ReorderBufferStreamCommit()
|
static |
Definition at line 1984 of file reorderbuffer.c.
1985{
1986 /* we should only call this for previously streamed transactions */
1988
1990
1992 {
1993 /*
1994 * Note, we send stream prepare even if a concurrent abort is
1995 * detected. See DecodePrepare for more information.
1996 */
2000
2001 /*
2002 * This is a PREPARED transaction, part of a two-phase commit. The
2003 * full cleanup will happen as part of the COMMIT PREPAREDs, so now
2004 * just truncate txn by removing changes and tuplecids.
2005 */
2007 /* Reset the CheckXidAlive */
2009 }
2010 else
2011 {
2014 }
2015}
References Assert, CheckXidAlive, fb(), ReorderBufferTXN::final_lsn, InvalidTransactionId, rbtxn_is_prepared, rbtxn_is_streamed, RBTXN_SENT_PREPARE, rbtxn_sent_prepare, ReorderBufferCleanupTXN(), ReorderBufferStreamTXN(), ReorderBufferTruncateTXN(), and ReorderBufferTXN::txn_flags.
Referenced by ReorderBufferReplay().
◆ ReorderBufferStreamTXN()
|
static |
Definition at line 4335 of file reorderbuffer.c.
4336{
4337 Snapshot snapshot_now;
4338 CommandId command_id;
4339 Size stream_bytes;
4341
4342 /* We can never reach here for a subtransaction. */
4344
4345 /*
4346 * We can't make any assumptions about base snapshot here, similar to what
4347 * ReorderBufferCommit() does. That relies on base_snapshot getting
4348 * transferred from subxact in ReorderBufferCommitChild(), but that was
4349 * not yet called as the transaction is in-progress.
4350 *
4351 * So just walk the subxacts and use the same logic here. But we only need
4352 * to do that once, when the transaction is streamed for the first time.
4353 * After that we need to reuse the snapshot from the previous run.
4354 *
4355 * Unlike DecodeCommit which adds xids of all the subtransactions in
4356 * snapshot's xip array via SnapBuildCommitTxn, we can't do that here but
4357 * we do add them to subxip array instead via ReorderBufferCopySnap. This
4358 * allows the catalog changes made in subtransactions decoded till now to
4359 * be visible.
4360 */
4362 {
4364
4365 /* make sure this transaction is streamed for the first time */
4367
4368 /* at the beginning we should have invalid command ID */
4370
4372 {
4374
4377 }
4378
4379 /*
4380 * If this transaction has no snapshot, it didn't make any changes to
4381 * the database till now, so there's nothing to decode.
4382 */
4384 {
4386 return;
4387 }
4388
4389 command_id = FirstCommandId;
4391 txn, command_id);
4392 }
4393 else
4394 {
4395 /* the transaction must have been already streamed */
4397
4398 /*
4399 * Nah, we already have snapshot from the previous streaming run. We
4400 * assume new subxacts can't move the LSN backwards, and so can't beat
4401 * the LSN condition in the previous branch (so no need to walk
4402 * through subxacts again). In fact, we must not do that as we may be
4403 * using snapshot half-way through the subxact.
4404 */
4405 command_id = txn->command_id;
4406
4407 /*
4408 * We can't use txn->snapshot_now directly because after the last
4409 * streaming run, we might have got some new sub-transactions. So we
4410 * need to add them to the snapshot.
4411 */
4413 txn, command_id);
4414
4415 /* Free the previously copied snapshot. */
4419 }
4420
4421 /*
4422 * Remember this information to be used later to update stats. We can't
4423 * update the stats here as an error while processing the changes would
4424 * lead to the accumulation of stats even though we haven't streamed all
4425 * the changes.
4426 */
4428 stream_bytes = txn->total_size;
4429
4430 /* Process and send the changes to output plugin. */
4432 command_id, true);
4433
4434 rb->streamCount += 1;
4435 rb->streamBytes += stream_bytes;
4436
4437 /* Don't consider already streamed transaction. */
4439
4440 /* update the decoding stats */
4442
4446}
References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::changes, ReorderBufferTXN::command_id, SnapshotData::copied, dlist_container, dlist_foreach, dlist_is_empty(), fb(), FirstCommandId, InvalidCommandId, InvalidXLogRecPtr, ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, ReorderBufferTXN::ninvalidations, rbtxn_is_streamed, rbtxn_is_toptxn, ReorderBufferCopySnap(), ReorderBufferFreeSnap(), ReorderBufferProcessTXN(), ReorderBufferTransferSnapToParent(), ReorderBufferTXN::snapshot_now, ReorderBufferTXN::subtxns, ReorderBufferTXN::total_size, and UpdateDecodingStats().
Referenced by ReorderBufferCheckMemoryLimit(), ReorderBufferProcessPartialChange(), and ReorderBufferStreamCommit().
◆ ReorderBufferToastAppendChunk()
|
static |
Definition at line 4988 of file reorderbuffer.c.
4990{
4993 bool found;
4995 bool isnull;
4996 Pointer chunk;
4998 Oid chunk_id;
5000
5003
5005
5008 Assert(!isnull);
5010 Assert(!isnull);
5011
5014
5015 if (!found)
5016 {
5018 ent->num_chunks = 0;
5019 ent->last_chunk_seq = 0;
5020 ent->size = 0;
5023
5026 chunk_seq, chunk_id);
5027 }
5031
5033 Assert(!isnull);
5034
5035 /* calculate size so we can allocate the right size at once later */
5039 /* could happen due to heap_form_tuple doing its thing */
5041 else
5043
5046 ent->num_chunks++;
5048}
References Assert, ReorderBufferChange::data, DatumGetInt32(), DatumGetObjectId(), DatumGetPointer(), dlist_init(), dlist_push_tail(), elog, ERROR, fastgetattr(), fb(), HASH_ENTER, hash_search(), IsToastRelation(), ReorderBufferChange::newtuple, ReorderBufferChange::node, RelationGetDescr, ReorderBufferToastInitHash(), ReorderBufferTXN::toast_hash, ReorderBufferChange::tp, VARATT_IS_EXTENDED(), VARATT_IS_SHORT(), VARHDRSZ, VARHDRSZ_SHORT, VARSIZE(), and VARSIZE_SHORT().
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferToastInitHash()
|
static |
Definition at line 4968 of file reorderbuffer.c.
4969{
4971
4973
4979}
References Assert, fb(), HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, and ReorderBufferTXN::toast_hash.
Referenced by ReorderBufferToastAppendChunk().
◆ ReorderBufferToastReplace()
|
static |
Definition at line 5071 of file reorderbuffer.c.
5073{
5074 TupleDesc desc;
5077 bool *isnull;
5080 Relation toast_rel;
5082 MemoryContext oldcontext;
5085
5086 /* no toast tuples changed */
5088 return;
5089
5090 /*
5091 * We're going to modify the size of the change. So, to make sure the
5092 * accounting is correct we record the current change size and then after
5093 * re-computing the change we'll subtract the recorded size and then
5094 * re-add the new change size at the end. We don't immediately subtract
5095 * the old size because if there is any error before we add the new size,
5096 * we will release the changes and that will update the accounting info
5097 * (subtracting the size from the counters). And we don't want to
5098 * underflow there.
5099 */
5101
5103
5104 /* we should only have toast tuples in an INSERT or UPDATE */
5106
5107 desc = RelationGetDescr(relation);
5108
5113
5115
5116 /* should we allocate from stack instead? */
5120
5122
5124
5126 {
5130
5131 /* va_rawsize is the size of the original datum -- including header */
5132 varatt_external toast_pointer;
5135 varlena *reconstructed;
5138
5140 continue;
5141
5142 /* not a varlena datatype */
5144 continue;
5145
5146 /* no data */
5148 continue;
5149
5150 /* ok, we know we have a toast datum */
5152
5153 /* no need to do anything if the tuple isn't external */
5155 continue;
5156
5158
5159 /*
5160 * Check whether the toast tuple changed, replace if so.
5161 */
5164 &toast_pointer.va_valueid,
5165 HASH_FIND,
5166 NULL);
5168 continue;
5169
5170 new_datum =
5172
5174
5176
5177 ent->reconstructed = reconstructed;
5178
5179 /* stitch toast tuple back together from its parts */
5181 {
5185 Pointer chunk;
5186
5190
5194
5196 VARDATA(chunk),
5199 }
5201
5202 /* make sure its marked as compressed or not */
5205 else
5207
5209 redirect_pointer.pointer = reconstructed;
5210
5214
5216 }
5217
5218 /*
5219 * Build tuple in separate memory & copy tuple back into the tuplebuf
5220 * passed to the output plugin. We can't directly heap_fill_tuple() into
5221 * the tuplebuf because attrs[] will point back into the current content.
5222 */
5226
5229
5230 /*
5231 * free resources we won't further need, more persistent stuff will be
5232 * free'd in ReorderBufferToastReset().
5233 */
5234 RelationClose(toast_rel);
5237 {
5240 }
5243 pfree(isnull);
5244
5245 MemoryContextSwitchTo(oldcontext);
5246
5247 /* subtract the old change size */
5249 /* now add the change back, with the correct size */
5251 ReorderBufferChangeSize(change));
5252}
References Assert, CompactAttribute::attisdropped, CompactAttribute::attlen, ReorderBufferChange::data, DatumGetPointer(), dlist_container, dlist_foreach, elog, ERROR, fastgetattr(), fb(), free, HASH_FIND, hash_search(), heap_deform_tuple(), heap_form_tuple(), HEAPTUPLESIZE, INDIRECT_POINTER_SIZE, MaxHeapTupleSize, memcpy(), MemoryContextSwitchTo(), TupleDescData::natts, ReorderBufferChange::newtuple, palloc0(), palloc0_array, pfree(), PointerGetDatum, RelationData::rd_rel, RelationClose(), RelationGetDescr, RelationGetRelationName, RelationIdGetRelation(), RelationIsValid, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), SET_VARSIZE(), SET_VARSIZE_COMPRESSED(), SET_VARTAG_EXTERNAL(), ReorderBufferTXN::toast_hash, ReorderBufferChange::tp, TupleDescCompactAttr(), varatt_external::va_rawsize, varatt_external::va_valueid, VARATT_EXTERNAL_GET_EXTSIZE(), VARATT_EXTERNAL_GET_POINTER, VARATT_EXTERNAL_IS_COMPRESSED(), VARATT_IS_EXTERNAL(), VARATT_IS_SHORT(), VARDATA(), VARDATA_EXTERNAL(), VARHDRSZ, VARSIZE(), and VARTAG_INDIRECT.
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferToastReset()
|
static |
Definition at line 5258 of file reorderbuffer.c.
5259{
5262
5264 return;
5265
5266 /* sequentially walk over the hash and free everything */
5269 {
5271
5274
5276 {
5277 ReorderBufferChange *change =
5279
5282 }
5283 }
5284
5287}
References dlist_container, dlist_delete(), dlist_foreach_modify, fb(), hash_destroy(), hash_seq_init(), hash_seq_search(), ReorderBufferChange::node, pfree(), ReorderBufferFreeChange(), and ReorderBufferTXN::toast_hash.
Referenced by ReorderBufferCheckAndTruncateAbortedTXN(), ReorderBufferFreeTXN(), ReorderBufferProcessTXN(), and ReorderBufferResetTXN().
◆ ReorderBufferTransferSnapToParent()
|
static |
Definition at line 1166 of file reorderbuffer.c.
1168{
1170
1172 {
1175 {
1176 /*
1177 * If the toplevel transaction already has a base snapshot but
1178 * it's newer than the subxact's, purge it.
1179 */
1181 {
1184 }
1185
1186 /*
1187 * The snapshot is now the top transaction's; transfer it, and
1188 * adjust the list position of the top transaction in the list by
1189 * moving it to where the subtransaction is.
1190 */
1194 &txn->base_snapshot_node);
1195
1196 /*
1197 * The subtransaction doesn't have a snapshot anymore (so it
1198 * mustn't be in the list.)
1199 */
1203 }
1204 else
1205 {
1206 /* Base snap of toplevel is fine, so subxact's is not needed */
1211 }
1212 }
1213}
References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::base_snapshot_lsn, ReorderBufferTXN::base_snapshot_node, dlist_delete(), dlist_insert_before(), fb(), InvalidXLogRecPtr, SnapBuildSnapDecRefcount(), and ReorderBufferTXN::xid.
Referenced by ReorderBufferAssignChild(), and ReorderBufferStreamTXN().
◆ ReorderBufferTruncateTXN()
|
static |
Definition at line 1657 of file reorderbuffer.c.
1658{
1659 dlist_mutable_iter iter;
1661
1662 /* cleanup subtransactions & their changes */
1664 {
1666
1668
1669 /*
1670 * Subtransactions are always associated to the toplevel TXN, even if
1671 * they originally were happening inside another subtxn, so we won't
1672 * ever recurse more than one level deep here.
1673 */
1676
1679 }
1680
1681 /* cleanup changes in the txn */
1683 {
1684 ReorderBufferChange *change;
1685
1687
1688 /* Check we're not mixing changes from different transactions. */
1690
1691 /* remove the change from its containing list */
1693
1694 /*
1695 * Instead of updating the memory counter for individual changes, we
1696 * sum up the size of memory to free so we can update the memory
1697 * counter all together below. This saves costs of maintaining the
1698 * max-heap.
1699 */
1701
1703 }
1704
1705 /* Update the memory counter */
1707
1709 {
1710 /*
1711 * If this is a prepared txn, cleanup the tuplecids we stored for
1712 * decoding catalog snapshot access. They are always stored in the
1713 * toplevel transaction.
1714 */
1716 {
1717 ReorderBufferChange *change;
1718
1720
1721 /* Check we're not mixing changes from different transactions. */
1724
1725 /* Remove the change from its containing list. */
1727
1729 }
1730 }
1731
1732 /*
1733 * Destroy the (relfilelocator, ctid) hashtable, so that we don't leak any
1734 * memory. We could also keep the hash table and update it with new ctid
1735 * values, but this seems simpler and good enough for now.
1736 */
1738 {
1741 }
1742
1743 /* If this txn is serialized then clean the disk space. */
1745 {
1748
1749 /*
1750 * We set this flag to indicate if the transaction is ever serialized.
1751 * We need this to accurately update the stats as otherwise the same
1752 * transaction can be counted as serialized multiple times.
1753 */
1755 }
1756
1757 /* also reset the number of entries in the transaction */
1758 txn->nentries_mem = 0;
1759 txn->nentries = 0;
1760}
References ReorderBufferChange::action, Assert, ReorderBufferTXN::changes, dlist_mutable_iter::cur, dlist_container, dlist_delete(), dlist_foreach_modify, fb(), hash_destroy(), ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, ReorderBufferChange::node, rbtxn_is_known_subxact, rbtxn_is_serialized, RBTXN_IS_SERIALIZED_CLEAR, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferFreeChange(), ReorderBufferMaybeMarkTXNStreamed(), ReorderBufferRestoreCleanup(), ReorderBufferTruncateTXN(), ReorderBufferTXN::subtxns, ReorderBufferTXN::tuplecid_hash, ReorderBufferTXN::tuplecids, ReorderBufferChange::txn, and ReorderBufferTXN::txn_flags.
Referenced by ReorderBufferCheckAndTruncateAbortedTXN(), ReorderBufferProcessTXN(), ReorderBufferResetTXN(), ReorderBufferStreamCommit(), and ReorderBufferTruncateTXN().
◆ ReorderBufferTXNByXid()
|
static |
Definition at line 654 of file reorderbuffer.c.
656{
657 ReorderBufferTXN *txn;
659 bool found;
660
662
663 /*
664 * Check the one-entry lookup cache first
665 */
667 rb->by_txn_last_xid == xid)
668 {
669 txn = rb->by_txn_last_txn;
670
672 {
673 /* found it, and it's valid */
676 return txn;
677 }
678
679 /*
680 * cached as non-existent, and asked not to create? Then nothing else
681 * to do.
682 */
683 if (!create)
685 /* otherwise fall through to create it */
686 }
687
688 /*
689 * If the cache wasn't hit or it yielded a "does-not-exist" and we want to
690 * create an entry.
691 */
692
693 /* search the lookup table */
696 &xid,
698 &found);
699 if (found)
700 txn = ent->txn;
701 else if (create)
702 {
703 /* initialize the new entry, if creation was requested */
706
708 ent->txn->xid = xid;
709 txn = ent->txn;
710 txn->first_lsn = lsn;
712
714 {
717 }
718 }
719 else
721
722 /* update cache */
723 rb->by_txn_last_xid = xid;
724 rb->by_txn_last_txn = txn;
725
727 *is_new = !found;
728
730 return txn;
731}
References Assert, AssertTXNLsnOrder(), dlist_push_tail(), fb(), ReorderBufferTXN::first_lsn, HASH_ENTER, HASH_FIND, hash_search(), ReorderBufferTXN::node, ReorderBufferAllocTXN(), ReorderBufferTXN::restart_decoding_lsn, TransactionIdIsValid, and XLogRecPtrIsValid.
Referenced by ReorderBufferAbort(), ReorderBufferAddDistributedInvalidations(), ReorderBufferAddInvalidations(), ReorderBufferAddNewTupleCids(), ReorderBufferAssignChild(), ReorderBufferCommit(), ReorderBufferCommitChild(), ReorderBufferFinishPrepared(), ReorderBufferForget(), ReorderBufferGetInvalidations(), ReorderBufferInvalidate(), ReorderBufferPrepare(), ReorderBufferProcessXid(), ReorderBufferQueueChange(), ReorderBufferQueueMessage(), ReorderBufferRememberPrepareInfo(), ReorderBufferSetBaseSnapshot(), ReorderBufferSkipPrepare(), ReorderBufferXidHasBaseSnapshot(), ReorderBufferXidHasCatalogChanges(), and ReorderBufferXidSetCatalogChanges().
◆ ReorderBufferTXNSizeCompare()
|
static |
Definition at line 3782 of file reorderbuffer.c.
3783{
3786
3788 return -1;
3790 return 1;
3791 return 0;
3792}
References a, b, fb(), and pairingheap_const_container.
Referenced by ReorderBufferAllocate().
◆ ReorderBufferXidHasBaseSnapshot()
| bool ReorderBufferXidHasBaseSnapshot | ( | ReorderBuffer * | rb, |
| TransactionId | xid | ||
| ) |
Definition at line 3735 of file reorderbuffer.c.
3736{
3737 ReorderBufferTXN *txn;
3738
3741
3742 /* transaction isn't known yet, ergo no snapshot */
3744 return false;
3745
3746 /* a known subtxn? operate on top-level txn instead */
3750
3752}
References ReorderBufferTXN::base_snapshot, fb(), InvalidXLogRecPtr, rbtxn_is_known_subxact, ReorderBufferTXNByXid(), and ReorderBufferTXN::toplevel_xid.
Referenced by SnapBuildCommitTxn(), SnapBuildDistributeSnapshotAndInval(), and SnapBuildProcessChange().
◆ ReorderBufferXidHasCatalogChanges()
| bool ReorderBufferXidHasCatalogChanges | ( | ReorderBuffer * | rb, |
| TransactionId | xid | ||
| ) |
Definition at line 3718 of file reorderbuffer.c.
References fb(), InvalidXLogRecPtr, rbtxn_has_catalog_changes, and ReorderBufferTXNByXid().
Referenced by SnapBuildXidHasCatalogChanges().
◆ ReorderBufferXidSetCatalogChanges()
| void ReorderBufferXidSetCatalogChanges | ( | ReorderBuffer * | rb, |
| TransactionId | xid, | ||
| XLogRecPtr | lsn | ||
| ) |
Definition at line 3646 of file reorderbuffer.c.
3648{
3649 ReorderBufferTXN *txn;
3650
3652
3654 {
3657 }
3658
3659 /*
3660 * Mark top-level transaction as having catalog changes too if one of its
3661 * children has so that the ReorderBufferBuildTupleCidHash can
3662 * conveniently check just top-level transaction and decide whether to
3663 * build the hash table or not.
3664 */
3666 {
3668
3670 {
3673 }
3674 }
3675}
References ReorderBufferTXN::catchange_node, dclist_push_tail(), fb(), rbtxn_get_toptxn, RBTXN_HAS_CATALOG_CHANGES, rbtxn_has_catalog_changes, rbtxn_is_subtxn, ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.
Referenced by SnapBuildProcessNewCid(), and xact_decode().
◆ ResolveCminCmaxDuringDecoding()
| bool ResolveCminCmaxDuringDecoding | ( | HTAB * | tuplecid_data, |
| Snapshot | snapshot, | ||
| HeapTuple | htup, | ||
| Buffer | buffer, | ||
| CommandId * | cmin, | ||
| CommandId * | cmax | ||
| ) |
Definition at line 5550 of file reorderbuffer.c.
5554{
5557 ForkNumber forkno;
5558 BlockNumber blockno;
5560
5561 /*
5562 * Return unresolved if tuplecid_data is not valid. That's because when
5563 * streaming in-progress transactions we may run into tuples with the CID
5564 * before actually decoding them. Think e.g. about INSERT followed by
5565 * TRUNCATE, where the TRUNCATE may not be decoded yet when applying the
5566 * INSERT. So in such cases, we assume the CID is from the future
5567 * command.
5568 */
5570 return false;
5571
5572 /* be careful about padding */
5574
5576
5577 /*
5578 * get relfilelocator from the buffer, no convenient way to access it
5579 * other than that.
5580 */
5582
5583 /* tuples can only be in the main fork */
5586
5588 &key.tid);
5589
5590restart:
5593
5594 /*
5595 * failed to find a mapping, check whether the table was rewritten and
5596 * apply mapping if so, but only do that once - there can be no new
5597 * mappings while we are in here since we have to hold a lock on the
5598 * relation.
5599 */
5601 {
5603 /* now check but don't update for a mapping again */
5605 goto restart;
5606 }
5608 return false;
5609
5610 if (cmin)
5611 *cmin = ent->cmin;
5612 if (cmax)
5613 *cmax = ent->cmax;
5614 return true;
5615}
References Assert, BufferGetTag(), BufferIsLocal, fb(), HASH_FIND, hash_search(), ItemPointerCopy(), ItemPointerGetBlockNumber(), MAIN_FORKNUM, HeapTupleData::t_self, HeapTupleData::t_tableOid, tuplecid_data, and UpdateLogicalMappings().
Referenced by HeapTupleSatisfiesHistoricMVCC().
◆ SetupCheckXidLive()
|
inlinestatic |
Definition at line 2050 of file reorderbuffer.c.
2051{
2052 /*
2053 * If the input transaction id is already set as a CheckXidAlive then
2054 * nothing to do.
2055 */
2057 return;
2058
2059 /*
2060 * setup CheckXidAlive if it's not committed yet. We don't check if the
2061 * xid is aborted. That will happen during catalog access.
2062 */
2064 CheckXidAlive = xid;
2065 else
2067}
References CheckXidAlive, InvalidTransactionId, TransactionIdDidCommit(), and TransactionIdEquals.
Referenced by ReorderBufferProcessTXN().
◆ StartupReorderBuffer()
Definition at line 4934 of file reorderbuffer.c.
4935{
4938
4941 {
4944 continue;
4945
4946 /* if it cannot be a slot, skip the directory */
4948 continue;
4949
4950 /*
4951 * ok, has to be a surviving logical slot, iterate and delete
4952 * everything starting with xid-*
4953 */
4955 }
4957}
References AllocateDir(), DEBUG2, fb(), FreeDir(), PG_REPLSLOT_DIR, ReadDir(), ReorderBufferCleanupSerializedTXNs(), and ReplicationSlotValidateName().
Referenced by StartupXLOG().
◆ TransactionIdInArray()
|
static |
Definition at line 5449 of file reorderbuffer.c.
References fb(), and xidComparator().
Referenced by UpdateLogicalMappings().
◆ UpdateLogicalMappings()
Definition at line 5472 of file reorderbuffer.c.
5473{
5477 ListCell *file;
5479
5482 {
5489 f_lo;
5490 RewriteMappingFile *f;
5491
5494 continue;
5495
5496 /* Ignore files that aren't ours */
5498 continue;
5499
5504
5506
5507 /* mapping for another database */
5509 continue;
5510
5511 /* mapping for another relation */
5513 continue;
5514
5515 /* did the creating transaction abort? */
5517 continue;
5518
5519 /* not for our transaction */
5521 continue;
5522
5523 /* ok, relevant, queue for apply */
5527 files = lappend(files, f);
5528 }
5530
5531 /* sort files so we apply them in LSN order */
5533
5534 foreach(file, files)
5535 {
5537
5539 snapshot->subxip[0]);
5541 pfree(f);
5542 }
5543}
References AllocateDir(), ApplyLogicalMappingFile(), DEBUG1, elog, ERROR, fb(), file_sort_by_lsn(), RewriteMappingFile::fname, FreeDir(), InvalidOid, IsSharedRelation(), lappend(), lfirst, list_sort(), LOGICAL_REWRITE_FORMAT, RewriteMappingFile::lsn, MyDatabaseId, NIL, palloc_object, pfree(), PG_LOGICAL_MAPPINGS_DIR, ReadDir(), SnapshotData::subxcnt, SnapshotData::subxip, TransactionIdDidCommit(), TransactionIdInArray(), and tuplecid_data.
Referenced by ResolveCminCmaxDuringDecoding().
Variable Documentation
◆ debug_logical_replication_streaming
| int debug_logical_replication_streaming = DEBUG_LOGICAL_REP_STREAMING_BUFFERED |
Definition at line 230 of file reorderbuffer.c.
Referenced by pa_send_data(), ReorderBufferCheckAndTruncateAbortedTXN(), and ReorderBufferCheckMemoryLimit().
◆ logical_decoding_work_mem
| int logical_decoding_work_mem |
Definition at line 226 of file reorderbuffer.c.
Referenced by ReorderBufferCheckMemoryLimit().
◆ max_changes_in_memory
Definition at line 227 of file reorderbuffer.c.
Referenced by ReorderBufferRestoreChanges().
