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 int32 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 int32 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,
2170{
2171 /* Discard the changes that we just streamed */
2173
2174 /* Free all resources allocated for toast reconstruction */
2176
2177 /* Return the spec insert change if it is not NULL */
2179 {
2182 }
2183
2184 /*
2185 * For the streaming case, stop the stream and remember the command ID and
2186 * snapshot for the streaming run.
2187 */
2189 {
2192 }
2193
2194 /* All changes must be deallocated */
2196}
2197
2198/*
2199 * Helper function for ReorderBufferReplay and ReorderBufferStreamTXN.
2200 *
2201 * Send data of a transaction (and its subtransactions) to the
2202 * output plugin. We iterate over the top and subtransactions (using a k-way
2203 * merge) and replay the changes in lsn order.
2204 *
2205 * If streaming is true then data will be sent using stream API.
2206 *
2207 * Note: "volatile" markers on some parameters are to avoid trouble with
2208 * PG_TRY inside the function.
2209 */
2210static void
2212 XLogRecPtr commit_lsn,
2215 bool streaming)
2216{
2225
2226 /* build data to be able to lookup the CommandIds of catalog tuples */
2228
2229 /* setup the initial snapshot */
2231
2232 /*
2233 * Decoding needs access to syscaches et al., which in turn use
2234 * heavyweight locks and such. Thus we need to have enough state around to
2235 * keep track of those. The easiest way is to simply use a transaction
2236 * internally. That also allows us to easily enforce that nothing writes
2237 * to the database by checking for xid assignments.
2238 *
2239 * When we're called via the SQL SRF there's already a transaction
2240 * started, so start an explicit subtransaction there.
2241 */
2243
2244 PG_TRY();
2245 {
2246 ReorderBufferChange *change;
2248 * changes */
2249
2252 else
2253 StartTransactionCommand();
2254
2255 /*
2256 * We only need to send begin/begin-prepare for non-streamed
2257 * transactions.
2258 */
2259 if (!streaming)
2260 {
2263 else
2265 }
2266
2269 {
2271 Oid reloid;
2272
2273 CHECK_FOR_INTERRUPTS();
2274
2275 /*
2276 * We can't call start stream callback before processing first
2277 * change.
2278 */
2280 {
2281 if (streaming)
2282 {
2286 }
2287 }
2288
2289 /*
2290 * Enforce correct ordering of changes, merged from multiple
2291 * subtransactions. The changes may have the same LSN due to
2292 * MULTI_INSERT xlog records.
2293 */
2295
2297
2298 /*
2299 * Set the current xid to detect concurrent aborts. This is
2300 * required for the cases when we decode the changes before the
2301 * COMMIT record is processed.
2302 */
2304 {
2307 }
2308
2310 {
2312
2313 /*
2314 * Confirmation for speculative insertion arrived. Simply
2315 * use as a normal record. It'll be cleaned up at the end
2316 * of INSERT processing.
2317 */
2321 change = specinsert;
2323
2324 /* intentionally fall through */
2325 pg_fallthrough;
2329 Assert(snapshot_now);
2330
2333
2334 /*
2335 * Mapped catalog tuple without data, emitted while
2336 * catalog table was in the process of being rewritten. We
2337 * can fail to look up the relfilenumber, because the
2338 * relmapper has no "historic" view, in contrast to the
2339 * normal catalog during decoding. Thus repeated rewrites
2340 * can cause a lookup failure. That's OK because we do not
2341 * decode catalog changes anyway. Normally such tuples
2342 * would be skipped over below, but we can't identify
2343 * whether the table should be logically logged without
2344 * mapping the relfilenumber to the oid.
2345 */
2353 MAIN_FORKNUM).str);
2354
2355 relation = RelationIdGetRelation(reloid);
2356
2359 reloid,
2361 MAIN_FORKNUM).str);
2362
2365
2366 /*
2367 * Ignore temporary heaps created during DDL unless the
2368 * plugin has asked for them.
2369 */
2372
2373 /*
2374 * For now ignore sequence changes entirely. Most of the
2375 * time they don't log changes using records we
2376 * understand, so it doesn't make sense to handle the few
2377 * cases we do.
2378 */
2381
2382 /* user-triggered change */
2384 {
2387 streaming);
2388
2389 /*
2390 * Only clear reassembled toast chunks if we're sure
2391 * they're not required anymore. The creator of the
2392 * tuple tells us.
2393 */
2396 }
2397 /* we're not interested in toast deletions */
2399 {
2400 /*
2401 * Need to reassemble the full toasted Datum in
2402 * memory, to ensure the chunks don't get reused till
2403 * we're done remove it from the list of this
2404 * transaction's changes. Otherwise it will get
2405 * freed/reused while restoring spooled data from
2406 * disk.
2407 */
2409
2412 change);
2413 }
2414
2415 change_done:
2416
2417 /*
2418 * If speculative insertion was confirmed, the record
2419 * isn't needed anymore.
2420 */
2422 {
2425 }
2426
2428 {
2429 RelationClose(relation);
2430 relation = NULL;
2431 }
2432 break;
2433
2435
2436 /*
2437 * Speculative insertions are dealt with by delaying the
2438 * processing of the insert until the confirmation record
2439 * arrives. For that we simply unlink the record from the
2440 * chain, so it does not get freed/reused while restoring
2441 * spooled data from disk.
2442 *
2443 * This is safe in the face of concurrent catalog changes
2444 * because the relevant relation can't be changed between
2445 * speculative insertion and confirmation due to
2446 * CheckTableNotInUse() and locking.
2447 */
2448
2449 /* Previous speculative insertion must be aborted */
2451
2452 /* and memorize the pending insertion */
2454 specinsert = change;
2455 break;
2456
2458
2459 /*
2460 * Abort for speculative insertion arrived. So cleanup the
2461 * specinsert tuple and toast hash.
2462 *
2463 * Note that we get the spec abort change for each toast
2464 * entry but we need to perform the cleanup only the first
2465 * time we get it for the main table.
2466 */
2468 {
2469 /*
2470 * We must clean the toast hash before processing a
2471 * completely new tuple to avoid confusion about the
2472 * previous tuple's toast chunks.
2473 */
2476
2477 /* We don't need this record anymore. */
2480 }
2481 break;
2482
2484 {
2488 Relation *relations;
2489
2492 {
2494 Relation rel;
2495
2496 rel = RelationIdGetRelation(relid);
2497
2500
2502 continue;
2503
2504 relations[nrelations++] = rel;
2505 }
2506
2507 /* Apply the truncate. */
2509 relations, change,
2510 streaming);
2511
2514
2515 break;
2516 }
2517
2520 break;
2521
2523 /* Execute the invalidation messages locally */
2526 break;
2527
2529 /* get rid of the old */
2531
2533 {
2535 snapshot_now =
2537 txn, command_id);
2538 }
2539
2540 /*
2541 * Restored from disk, need to be careful not to double
2542 * free. We could introduce refcounting for that, but for
2543 * now this seems infrequent enough not to care.
2544 */
2546 {
2547 snapshot_now =
2549 txn, command_id);
2550 }
2551 else
2552 {
2554 }
2555
2556 /* and continue with the new one */
2558 break;
2559
2562
2564 {
2566
2568 {
2569 /* we don't use the global one anymore */
2571 txn, command_id);
2572 }
2573
2574 snapshot_now->curcid = command_id;
2575
2578 }
2579
2580 break;
2581
2584 break;
2585 }
2586
2587 /*
2588 * It is possible that the data is not sent to downstream for a
2589 * long time either because the output plugin filtered it or there
2590 * is a DDL that generates a lot of data that is not processed by
2591 * the plugin. So, in such cases, the downstream can timeout. To
2592 * avoid that we try to send a keepalive message if required.
2593 * Trying to send a keepalive message after every change has some
2594 * overhead, but testing showed there is no noticeable overhead if
2595 * we do it after every ~100 changes.
2596 */
2597#define CHANGES_THRESHOLD 100
2598
2600 {
2602 changes_count = 0;
2603 }
2604 }
2605
2606 /* speculative insertion record must be freed by now */
2608
2609 /* clean up the iterator */
2612
2613 /*
2614 * Update total transaction count and total bytes processed by the
2615 * transaction and its subtransactions. Ensure to not count the
2616 * streamed transaction multiple times.
2617 *
2618 * Note that the statistics computation has to be done after
2619 * ReorderBufferIterTXNFinish as it releases the serialized change
2620 * which we have already accounted in ReorderBufferIterTXNNext.
2621 */
2623 rb->totalTxns++;
2624
2626
2627 /*
2628 * Done with current changes, send the last message for this set of
2629 * changes depending upon streaming mode.
2630 */
2631 if (streaming)
2632 {
2634 {
2637 }
2638 }
2639 else
2640 {
2641 /*
2642 * Call either PREPARE (for two-phase transactions) or COMMIT (for
2643 * regular ones).
2644 */
2646 {
2650 }
2651 else
2653 }
2654
2655 /* this is just a sanity check against bad output plugin behaviour */
2658 GetCurrentTransactionId());
2659
2660 /*
2661 * Remember the command ID and snapshot for the next set of changes in
2662 * streaming mode.
2663 */
2664 if (streaming)
2668
2669 /* cleanup */
2671
2672 /*
2673 * Aborting the current (sub-)transaction as a whole has the right
2674 * semantics. We want all locks acquired in here to be released, not
2675 * reassigned to the parent and we do not want any database access
2676 * have persistent effects.
2677 */
2678 AbortCurrentTransaction();
2679
2680 /* make sure there's no cache pollution */
2682 {
2684 InvalidateSystemCaches();
2685 }
2686 else
2687 {
2690 txn->invalidations_distributed);
2691 }
2692
2694 {
2697 CurrentResourceOwner = cowner;
2698 }
2699
2700 /*
2701 * We are here due to one of the four reasons: 1. Decoding an
2702 * in-progress txn. 2. Decoding a prepared txn. 3. Decoding of a
2703 * prepared txn that was (partially) streamed. 4. Decoding a committed
2704 * txn.
2705 *
2706 * For 1, we allow truncation of txn data by removing the changes
2707 * already streamed but still keeping other things like invalidations,
2708 * snapshot, and tuplecids. For 2 and 3, we indicate
2709 * ReorderBufferTruncateTXN to do more elaborate truncation of txn
2710 * data as the entire transaction has been decoded except for commit.
2711 * For 4, as the entire txn has been decoded, we can fully clean up
2712 * the TXN reorder buffer.
2713 */
2715 {
2716 if (streaming)
2718
2720 /* Reset the CheckXidAlive */
2722 }
2723 else
2725 }
2726 PG_CATCH();
2727 {
2730
2731 /* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */
2734
2736
2737 /*
2738 * Force cache invalidation to happen outside of a valid transaction
2739 * to prevent catalog access as we just caught an error.
2740 */
2741 AbortCurrentTransaction();
2742
2743 /* make sure there's no cache pollution */
2745 {
2747 InvalidateSystemCaches();
2748 }
2749 else
2750 {
2753 txn->invalidations_distributed);
2754 }
2755
2757 {
2760 CurrentResourceOwner = cowner;
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 specinsert);
2798 }
2799 else
2800 {
2803 PG_RE_THROW();
2804 }
2805 }
2806 PG_END_TRY();
2807}
2808
2809/*
2810 * Perform the replay of a transaction and its non-aborted subtransactions.
2811 *
2812 * Subtransactions previously have to be processed by
2813 * ReorderBufferCommitChild(), even if previously assigned to the toplevel
2814 * transaction with ReorderBufferAssignChild.
2815 *
2816 * This interface is called once a prepare or toplevel commit is read for both
2817 * streamed as well as non-streamed transactions.
2818 */
2819static void
2823 TimestampTz commit_time,
2825{
2826 Snapshot snapshot_now;
2828
2829 txn->final_lsn = commit_lsn;
2830 txn->end_lsn = end_lsn;
2831 txn->commit_time = commit_time;
2832 txn->origin_id = origin_id;
2833 txn->origin_lsn = origin_lsn;
2834
2835 /*
2836 * If the transaction was (partially) streamed, we need to commit it in a
2837 * 'streamed' way. That is, we first stream the remaining part of the
2838 * transaction, and then invoke stream_commit message.
2839 *
2840 * Called after everything (origin ID, LSN, ...) is stored in the
2841 * transaction to avoid passing that information directly.
2842 */
2844 {
2846 return;
2847 }
2848
2849 /*
2850 * If this transaction has no snapshot, it didn't make any changes to the
2851 * database, so there's nothing to decode. Note that
2852 * ReorderBufferCommitChild will have transferred any snapshots from
2853 * subtransactions if there were any.
2854 */
2856 {
2858
2859 /*
2860 * Removing this txn before a commit might result in the computation
2861 * of an incorrect restart_lsn. See SnapBuildProcessRunningXacts.
2862 */
2865 return;
2866 }
2867
2868 snapshot_now = txn->base_snapshot;
2869
2870 /* Process and send the changes to output plugin. */
2872 command_id, false);
2873}
2874
2875/*
2876 * Commit a transaction.
2877 *
2878 * See comments for ReorderBufferReplay().
2879 */
2880void
2883 TimestampTz commit_time,
2885{
2886 ReorderBufferTXN *txn;
2887
2889 false);
2890
2891 /* unknown transaction, nothing to replay */
2893 return;
2894
2896 origin_id, origin_lsn);
2897}
2898
2899/*
2900 * Record the prepare information for a transaction. Also, mark the transaction
2901 * as a prepared transaction.
2902 */
2903bool
2906 TimestampTz prepare_time,
2908{
2909 ReorderBufferTXN *txn;
2910
2912
2913 /* unknown transaction, nothing to do */
2915 return false;
2916
2917 /*
2918 * Remember the prepare information to be later used by commit prepared in
2919 * case we skip doing prepare.
2920 */
2921 txn->final_lsn = prepare_lsn;
2922 txn->end_lsn = end_lsn;
2923 txn->prepare_time = prepare_time;
2924 txn->origin_id = origin_id;
2925 txn->origin_lsn = origin_lsn;
2926
2927 /* Mark this transaction as a prepared transaction */
2930
2931 return true;
2932}
2933
2934/* Remember that we have skipped prepare */
2935void
2937{
2938 ReorderBufferTXN *txn;
2939
2941
2942 /* unknown transaction, nothing to do */
2944 return;
2945
2946 /* txn must have been marked as a prepared transaction */
2949}
2950
2951/*
2952 * Prepare a two-phase transaction.
2953 *
2954 * See comments for ReorderBufferReplay().
2955 */
2956void
2958 char *gid)
2959{
2960 ReorderBufferTXN *txn;
2961
2963 false);
2964
2965 /* unknown transaction, nothing to replay */
2967 return;
2968
2969 /*
2970 * txn must have been marked as a prepared transaction and must have
2971 * neither been skipped nor sent a prepare. Also, the prepare info must
2972 * have been updated in it by now.
2973 */
2976
2978
2981
2982 /*
2983 * Send a prepare if not already done so. This might occur if we have
2984 * detected a concurrent abort while replaying the non-streaming
2985 * transaction.
2986 */
2988 {
2991 }
2992}
2993
2994/*
2995 * This is used to handle COMMIT/ROLLBACK PREPARED.
2996 */
2997void
3000 XLogRecPtr two_phase_at,
3003{
3004 ReorderBufferTXN *txn;
3005 XLogRecPtr prepare_end_lsn;
3006 TimestampTz prepare_time;
3007
3009
3010 /* unknown transaction, nothing to do */
3012 return;
3013
3014 /*
3015 * By this time the txn has the prepare record information, remember it to
3016 * be later used for rollback.
3017 */
3018 prepare_end_lsn = txn->end_lsn;
3019 prepare_time = txn->prepare_time;
3020
3021 /* add the gid in the txn */
3023
3024 /*
3025 * It is possible that this transaction is not decoded at prepare time
3026 * either because by that time we didn't have a consistent snapshot, or
3027 * two_phase was not enabled, or it was decoded earlier but we have
3028 * restarted. We only need to send the prepare if it was not decoded
3029 * earlier. We don't need to decode the xact for aborts if it is not done
3030 * already.
3031 */
3033 {
3034 /*
3035 * txn must have been marked as a prepared transaction and skipped but
3036 * not sent a prepare. Also, the prepare info must have been updated
3037 * in txn even if we skip prepare.
3038 */
3042
3043 /*
3044 * By this time the txn has the prepare record information and it is
3045 * important to use that so that downstream gets the accurate
3046 * information. If instead, we have passed commit information here
3047 * then downstream can behave as it has already replayed commit
3048 * prepared after the restart.
3049 */
3052 }
3053
3054 txn->final_lsn = commit_lsn;
3055 txn->end_lsn = end_lsn;
3056 txn->commit_time = commit_time;
3057 txn->origin_id = origin_id;
3058 txn->origin_lsn = origin_lsn;
3059
3062 else
3064
3065 /* cleanup: make sure there's no cache pollution */
3067 txn->invalidations);
3069}
3070
3071/*
3072 * Abort a transaction that possibly has previous changes. Needs to be first
3073 * called for subtransactions and then for the toplevel xid.
3074 *
3075 * NB: Transactions handled here have to have actively aborted (i.e. have
3076 * produced an abort record). Implicitly aborted transactions are handled via
3077 * ReorderBufferAbortOld(); transactions we're just not interested in, but
3078 * which have committed are handled in ReorderBufferForget().
3079 *
3080 * This function purges this transaction and its contents from memory and
3081 * disk.
3082 */
3083void
3085 TimestampTz abort_time)
3086{
3087 ReorderBufferTXN *txn;
3088
3090 false);
3091
3092 /* unknown, nothing to remove */
3094 return;
3095
3096 txn->abort_time = abort_time;
3097
3098 /* For streamed transactions notify the remote node about the abort. */
3100 {
3102
3103 /*
3104 * We might have decoded changes for this transaction that could load
3105 * the cache as per the current transaction's view (consider DDL's
3106 * happened in this transaction). We don't want the decoding of future
3107 * transactions to use those cache entries so execute only the inval
3108 * messages in this transaction.
3109 */
3112 txn->invalidations);
3113 }
3114
3115 /* cosmetic... */
3116 txn->final_lsn = lsn;
3117
3118 /* remove potential on-disk data, and deallocate */
3120}
3121
3122/*
3123 * Abort all transactions that aren't actually running anymore because the
3124 * server restarted.
3125 *
3126 * NB: These really have to be transactions that have aborted due to a server
3127 * crash/immediate restart, as we don't deal with invalidations here.
3128 */
3129void
3131{
3133
3134 /*
3135 * Iterate through all (potential) toplevel TXNs and abort all that are
3136 * older than what possibly can be running. Once we've found the first
3137 * that is alive we stop, there might be some that acquired an xid earlier
3138 * but started writing later, but it's unlikely and they will be cleaned
3139 * up in a later call to this function.
3140 */
3142 {
3143 ReorderBufferTXN *txn;
3144
3146
3148 {
3150
3151 /* Notify the remote node about the crash/immediate restart. */
3154
3155 /* remove potential on-disk data, and deallocate this tx */
3157 }
3158 else
3159 return;
3160 }
3161}
3162
3163/*
3164 * Forget the contents of a transaction if we aren't interested in its
3165 * contents. Needs to be first called for subtransactions and then for the
3166 * toplevel xid.
3167 *
3168 * This is significantly different to ReorderBufferAbort() because
3169 * transactions that have committed need to be treated differently from aborted
3170 * ones since they may have modified the catalog.
3171 *
3172 * Note that this is only allowed to be called in the moment a transaction
3173 * commit has just been read, not earlier; otherwise later records referring
3174 * to this xid might re-create the transaction incompletely.
3175 */
3176void
3178{
3179 ReorderBufferTXN *txn;
3180
3182 false);
3183
3184 /* unknown, nothing to forget */
3186 return;
3187
3188 /* this transaction mustn't be streamed */
3190
3191 /* cosmetic... */
3192 txn->final_lsn = lsn;
3193
3194 /*
3195 * Process only cache invalidation messages in this transaction if there
3196 * are any. Even if we're not interested in the transaction's contents, it
3197 * could have manipulated the catalog and we need to update the caches
3198 * according to that.
3199 */
3202 txn->invalidations);
3203 else
3205
3206 /* remove potential on-disk data, and deallocate */
3208}
3209
3210/*
3211 * Invalidate cache for those transactions that need to be skipped just in case
3212 * catalogs were manipulated as part of the transaction.
3213 *
3214 * Note that this is a special-purpose function for prepared transactions where
3215 * we don't want to clean up the TXN even when we decide to skip it. See
3216 * DecodePrepare.
3217 */
3218void
3220{
3221 ReorderBufferTXN *txn;
3222
3224 false);
3225
3226 /* unknown, nothing to do */
3228 return;
3229
3230 /*
3231 * Process cache invalidation messages if there are any. Even if we're not
3232 * interested in the transaction's contents, it could have manipulated the
3233 * catalog and we need to update the caches according to that.
3234 */
3237 txn->invalidations);
3238 else
3240}
3241
3242
3243/*
3244 * Execute invalidations happening outside the context of a decoded
3245 * transaction. That currently happens either for xid-less commits
3246 * (cf. RecordTransactionCommit()) or for invalidations in uninteresting
3247 * transactions (via ReorderBufferForget()).
3248 */
3249void
3251 SharedInvalidationMessage *invalidations)
3252{
3257
3260
3261 /*
3262 * Force invalidations to happen outside of a valid transaction - that way
3263 * entries will just be marked as invalid without accessing the catalog.
3264 * That's advantageous because we don't need to setup the full state
3265 * necessary for catalog access.
3266 */
3268 AbortCurrentTransaction();
3269
3272
3274 {
3277 CurrentResourceOwner = cowner;
3278 }
3279}
3280
3281/*
3282 * Tell reorderbuffer about an xid seen in the WAL stream. Has to be called at
3283 * least once for every xid in XLogRecord->xl_xid (other places in records
3284 * may, but do not have to be passed through here).
3285 *
3286 * Reorderbuffer keeps some data structures about transactions in LSN order,
3287 * for efficiency. To do that it has to know about when transactions are seen
3288 * first in the WAL. As many types of records are not actually interesting for
3289 * logical decoding, they do not necessarily pass through here.
3290 */
3291void
3293{
3294 /* many records won't have an xid assigned, centralize check here */
3297}
3298
3299/*
3300 * Add a new snapshot to this transaction that may only used after lsn 'lsn'
3301 * because the previous snapshot doesn't describe the catalog correctly for
3302 * following rows.
3303 */
3304void
3307{
3309
3312
3314}
3315
3316/*
3317 * Set up the transaction's base snapshot.
3318 *
3319 * If we know that xid is a subtransaction, set the base snapshot on the
3320 * top-level transaction instead.
3321 */
3322void
3325{
3326 ReorderBufferTXN *txn;
3328
3330
3331 /*
3332 * Fetch the transaction to operate on. If we know it's a subtransaction,
3333 * operate on its top-level transaction instead.
3334 */
3340
3342 txn->base_snapshot_lsn = lsn;
3344
3346}
3347
3348/*
3349 * Access the catalog with this CommandId at this point in the changestream.
3350 *
3351 * May only be called for command ids > 1
3352 */
3353void
3356{
3358
3361
3363}
3364
3365/*
3366 * Update memory counters to account for the new or removed change.
3367 *
3368 * We update two counters - in the reorder buffer, and in the transaction
3369 * containing the change. The reorder buffer counter allows us to quickly
3370 * decide if we reached the memory limit, the transaction counter allows
3371 * us to quickly pick the largest transaction for eviction.
3372 *
3373 * Either txn or change must be non-NULL at least. We update the memory
3374 * counter of txn if it's non-NULL, otherwise change->txn.
3375 *
3376 * When streaming is enabled, we need to update the toplevel transaction
3377 * counters instead - we don't really care about subtransactions as we
3378 * can't stream them individually anyway, and we only pick toplevel
3379 * transactions for eviction. So only toplevel transactions matter.
3380 */
3381static void
3383 ReorderBufferChange *change,
3384 ReorderBufferTXN *txn,
3386{
3387 ReorderBufferTXN *toptxn;
3388
3389 Assert(txn || change);
3390
3391 /*
3392 * Ignore tuple CID changes, because those are not evicted when reaching
3393 * memory limit. So we just don't count them, because it might easily
3394 * trigger a pointless attempt to spill.
3395 */
3397 return;
3398
3400 return;
3401
3403 txn = change->txn;
3405
3406 /*
3407 * Update the total size in top level as well. This is later used to
3408 * compute the decoding stats.
3409 */
3410 toptxn = rbtxn_get_toptxn(txn);
3411
3413 {
3415
3418
3419 /* Update the total size in the top transaction. */
3421
3422 /* Update the max-heap */
3426 }
3427 else
3428 {
3432
3433 /* Update the total size in the top transaction. */
3435
3436 /* Update the max-heap */
3440 }
3441
3443}
3444
3445/*
3446 * Add new (relfilelocator, tid) -> (cmin, cmax) mappings.
3447 *
3448 * We do not include this change type in memory accounting, because we
3449 * keep CIDs in a separate list and do not evict them when reaching
3450 * the memory limit.
3451 */
3452void
3457{
3459 ReorderBufferTXN *txn;
3460
3462
3468 change->lsn = lsn;
3469 change->txn = txn;
3471
3473 txn->ntuplecids++;
3474}
3475
3476/*
3477 * Add new invalidation messages to the reorder buffer queue.
3478 */
3479static void
3482 SharedInvalidationMessage *msgs)
3483{
3484 ReorderBufferChange *change;
3485
3492
3494}
3495
3496/*
3497 * A helper function for ReorderBufferAddInvalidations() and
3498 * ReorderBufferAddDistributedInvalidations() to accumulate the invalidation
3499 * messages to the **invals_out.
3500 */
3501static void
3506{
3508 {
3512 }
3513 else
3514 {
3515 /* Enlarge the array of inval messages */
3516 *invals_out =
3522 }
3523}
3524
3525/*
3526 * Accumulate the invalidations for executing them later.
3527 *
3528 * This needs to be called for each XLOG_XACT_INVALIDATIONS message and
3529 * accumulates all the invalidation messages in the toplevel transaction, if
3530 * available, otherwise in the current transaction, as well as in the form of
3531 * change in reorder buffer. We require to record it in form of the change
3532 * so that we can execute only the required invalidations instead of executing
3533 * all the invalidations on each CommandId increment. We also need to
3534 * accumulate these in the txn buffer because in some cases where we skip
3535 * processing the transaction (see ReorderBufferForget), we need to execute
3536 * all the invalidations together.
3537 */
3538void
3541 SharedInvalidationMessage *msgs)
3542{
3543 ReorderBufferTXN *txn;
3544 MemoryContext oldcontext;
3545
3547
3549
3550 /*
3551 * Collect all the invalidations under the top transaction, if available,
3552 * so that we can execute them all together. See comments atop this
3553 * function.
3554 */
3555 txn = rbtxn_get_toptxn(txn);
3556
3557 Assert(nmsgs > 0);
3558
3560 &txn->ninvalidations,
3561 msgs, nmsgs);
3562
3564
3565 MemoryContextSwitchTo(oldcontext);
3566}
3567
3568/*
3569 * Accumulate the invalidations distributed by other committed transactions
3570 * for executing them later.
3571 *
3572 * This function is similar to ReorderBufferAddInvalidations() but stores
3573 * the given inval messages to the txn->invalidations_distributed with the
3574 * overflow check.
3575 *
3576 * This needs to be called by committed transactions to distribute their
3577 * inval messages to in-progress transactions.
3578 */
3579void
3582 SharedInvalidationMessage *msgs)
3583{
3584 ReorderBufferTXN *txn;
3585 MemoryContext oldcontext;
3586
3588
3590
3591 /*
3592 * Collect all the invalidations under the top transaction, if available,
3593 * so that we can execute them all together. See comments
3594 * ReorderBufferAddInvalidations.
3595 */
3596 txn = rbtxn_get_toptxn(txn);
3597
3598 Assert(nmsgs > 0);
3599
3601 {
3602 /*
3603 * Check the transaction has enough space for storing distributed
3604 * invalidation messages.
3605 */
3607 {
3608 /*
3609 * Mark the invalidation message as overflowed and free up the
3610 * messages accumulated so far.
3611 */
3613
3615 {
3618 txn->ninvalidations_distributed = 0;
3619 }
3620 }
3621 else
3623 &txn->ninvalidations_distributed,
3624 msgs, nmsgs);
3625 }
3626
3627 /* Queue the invalidation messages into the transaction */
3629
3630 MemoryContextSwitchTo(oldcontext);
3631}
3632
3633/*
3634 * Apply all invalidations we know. Possibly we only need parts at this point
3635 * in the changestream but we don't know which those are.
3636 */
3637static void
3639{
3641
3644}
3645
3646/*
3647 * Mark a transaction as containing catalog changes
3648 */
3649void
3651 XLogRecPtr lsn)
3652{
3653 ReorderBufferTXN *txn;
3654
3656
3658 {
3661 }
3662
3663 /*
3664 * Mark top-level transaction as having catalog changes too if one of its
3665 * children has so that the ReorderBufferBuildTupleCidHash can
3666 * conveniently check just top-level transaction and decide whether to
3667 * build the hash table or not.
3668 */
3670 {
3672
3674 {
3677 }
3678 }
3679}
3680
3681/*
3682 * Return palloc'ed array of the transactions that have changed catalogs.
3683 * The returned array is sorted in xidComparator order.
3684 *
3685 * The caller must free the returned array when done with it.
3686 */
3687TransactionId *
3689{
3690 dlist_iter iter;
3692 size_t xcnt = 0;
3693
3694 /* Quick return if the list is empty */
3697
3698 /* Initialize XID array */
3701 {
3703 catchange_node,
3704 iter.cur);
3705
3707
3708 xids[xcnt++] = txn->xid;
3709 }
3710
3712
3714 return xids;
3715}
3716
3717/*
3718 * Query whether a transaction is already *known* to contain catalog
3719 * changes. This can be wrong until directly before the commit!
3720 */
3721bool
3723{
3724 ReorderBufferTXN *txn;
3725
3727 false);
3729 return false;
3730
3732}
3733
3734/*
3735 * ReorderBufferXidHasBaseSnapshot
3736 * Have we already set the base snapshot for the given txn/subtxn?
3737 */
3738bool
3740{
3741 ReorderBufferTXN *txn;
3742
3745
3746 /* transaction isn't known yet, ergo no snapshot */
3748 return false;
3749
3750 /* a known subtxn? operate on top-level txn instead */
3754
3756}
3757
3758
3759/*
3760 * ---------------------------------------
3761 * Disk serialization support
3762 * ---------------------------------------
3763 */
3764
3765/*
3766 * Ensure the IO buffer is >= sz.
3767 */
3768static void
3770{
3772 {
3775 }
3777 {
3780 }
3781}
3782
3783
3784/* Compare two transactions by size */
3785static int
3787{
3790
3792 return -1;
3794 return 1;
3795 return 0;
3796}
3797
3798/*
3799 * Find the largest transaction (toplevel or subxact) to evict (spill to disk).
3800 */
3803{
3805
3806 /* Get the largest transaction from the max-heap */
3809
3813
3815}
3816
3817/*
3818 * Find the largest streamable (and non-aborted) toplevel transaction to evict
3819 * (by streaming).
3820 *
3821 * This can be seen as an optimized version of ReorderBufferLargestTXN, which
3822 * should give us the same transaction (because we don't update memory account
3823 * for subtransaction with streaming, so it's always 0). But we can simply
3824 * iterate over the limited number of toplevel transactions that have a base
3825 * snapshot. There is no use of selecting a transaction that doesn't have base
3826 * snapshot because we don't decode such transactions. Also, we do not select
3827 * the transaction which doesn't have any streamable change.
3828 *
3829 * Note that, we skip transactions that contain incomplete changes. There
3830 * is a scope of optimization here such that we can select the largest
3831 * transaction which has incomplete changes. But that will make the code and
3832 * design quite complex and that might not be worth the benefit. If we plan to
3833 * stream the transactions that contain incomplete changes then we need to
3834 * find a way to partially stream/truncate the transaction changes in-memory
3835 * and build a mechanism to partially truncate the spilled files.
3836 * Additionally, whenever we partially stream the transaction we need to
3837 * maintain the last streamed lsn and next time we need to restore from that
3838 * segment and the offset in WAL. As we stream the changes from the top
3839 * transaction and restore them subtransaction wise, we need to even remember
3840 * the subxact from where we streamed the last change.
3841 */
3844{
3845 dlist_iter iter;
3848
3849 /* Find the largest top-level transaction having a base snapshot. */
3851 {
3852 ReorderBufferTXN *txn;
3853
3855
3856 /* must not be a subtxn */
3858 /* base_snapshot must be set */
3860
3861 /* Don't consider these kinds of transactions for eviction. */
3863 !rbtxn_has_streamable_change(txn) ||
3864 rbtxn_is_aborted(txn))
3865 continue;
3866
3867 /* Find the largest of the eviction candidates. */
3869 (txn->total_size > 0))
3870 {
3871 largest = txn;
3873 }
3874 }
3875
3877}
3878
3879/*
3880 * Check whether the logical_decoding_work_mem limit was reached, and if yes
3881 * pick the largest (sub)transaction at-a-time to evict and spill its changes to
3882 * disk or send to the output plugin until we reach under the memory limit.
3883 *
3884 * If debug_logical_replication_streaming is set to "immediate", stream or
3885 * serialize the changes immediately.
3886 *
3887 * XXX At this point we select the transactions until we reach under the memory
3888 * limit, but we might also adapt a more elaborate eviction strategy - for example
3889 * evicting enough transactions to free certain fraction (e.g. 50%) of the memory
3890 * limit.
3891 */
3892static void
3894{
3895 ReorderBufferTXN *txn;
3897
3899 {
3900 /*
3901 * Update the statistics as the memory usage has reached the limit. We
3902 * report the statistics update later in this function since we can
3903 * update the slot statistics altogether while streaming or
3904 * serializing transactions in most cases.
3905 */
3906 rb->memExceededCount += 1;
3907 }
3909 {
3910 /*
3911 * Bail out if debug_logical_replication_streaming is buffered and we
3912 * haven't exceeded the memory limit.
3913 */
3914 return;
3915 }
3916
3917 /*
3918 * If debug_logical_replication_streaming is immediate, loop until there's
3919 * no change. Otherwise, loop until we reach under the memory limit. One
3920 * might think that just by evicting the largest (sub)transaction we will
3921 * come under the memory limit based on assumption that the selected
3922 * transaction is at least as large as the most recent change (which
3923 * caused us to go over the memory limit). However, that is not true
3924 * because a user can reduce the logical_decoding_work_mem to a smaller
3925 * value before the most recent change.
3926 */
3929 rb->size > 0))
3930 {
3931 /*
3932 * Pick the largest non-aborted transaction and evict it from memory
3933 * by streaming, if possible. Otherwise, spill to disk.
3934 */
3937 {
3938 /* we know there has to be one, because the size is not zero */
3942
3943 /* skip the transaction if aborted */
3945 continue;
3946
3948 }
3949 else
3950 {
3951 /*
3952 * Pick the largest transaction (or subtransaction) and evict it
3953 * from memory by serializing it to disk.
3954 */
3956
3957 /* we know there has to be one, because the size is not zero */
3958 Assert(txn);
3961
3962 /* skip the transaction if aborted */
3964 continue;
3965
3967 }
3968
3969 /*
3970 * After eviction, the transaction should have no entries in memory,
3971 * and should use 0 bytes for changes.
3972 */
3975
3976 /*
3977 * We've reported the memExceededCount update while streaming or
3978 * serializing the transaction.
3979 */
3981 }
3982
3985
3986 /* We must be under the memory limit now. */
3988}
3989
3990/*
3991 * Spill data of a large transaction (and its subtransactions) to disk.
3992 */
3993static void
3995{
4000 Size spilled = 0;
4002
4005
4006 /* do the same to all child TXs */
4008 {
4010
4013 }
4014
4015 /* serialize changestream */
4017 {
4018 ReorderBufferChange *change;
4019
4021
4022 /*
4023 * store in segment in which it belongs by start lsn, don't split over
4024 * multiple segments tho
4025 */
4028 {
4030
4033
4035
4036 /*
4037 * No need to care about TLIs here, only used during a single run,
4038 * so each LSN only maps to a specific WAL record.
4039 */
4041 curOpenSegNo);
4042
4043 /* open segment, create it if necessary */
4046
4049 (errcode_for_file_access(),
4051 }
4052
4056
4057 spilled++;
4058 }
4059
4060 /* Update the memory counter */
4062
4063 /* update the statistics iff we have spilled anything */
4064 if (spilled)
4065 {
4066 rb->spillCount += 1;
4067 rb->spillBytes += size;
4068
4069 /* don't consider already serialized transactions */
4071
4072 /* update the decoding stats */
4074 }
4075
4078 txn->nentries_mem = 0;
4080
4083}
4084
4085/*
4086 * Serialize individual change to disk.
4087 */
4088static void
4091{
4092 ReorderBufferDiskChange *ondisk;
4094
4096
4099
4101 {
4102 /* fall through these, they're all similar enough */
4107 {
4110 newtup;
4113
4116
4118 {
4122 }
4123
4125 {
4129 }
4130
4131 /* make sure we have enough space */
4133
4135 /* might have been reallocated above */
4137
4139 {
4142
4145 }
4146
4148 {
4151
4154 }
4155 break;
4156 }
4158 {
4161
4165
4167
4168 /* might have been reallocated above */
4170
4171 /* write the prefix including the size */
4175 prefix_size);
4176 data += prefix_size;
4177
4178 /* write the message including the size */
4184
4185 break;
4186 }
4188 {
4192
4194
4197
4198 /* might have been reallocated above */
4202
4203 break;
4204 }
4206 {
4209
4211
4215
4216 /* make sure we have enough space */
4219 /* might have been reallocated above */
4221
4224
4226 {
4230 }
4231
4233 {
4237 }
4238 break;
4239 }
4241 {
4242 Size size;
4244
4245 /* account for the OIDs of truncated relations */
4247 sz += size;
4248
4249 /* make sure we have enough space */
4251
4253 /* might have been reallocated above */
4255
4257 data += size;
4258
4259 break;
4260 }
4265 /* ReorderBufferChange contains everything important */
4266 break;
4267 }
4268
4269 ondisk->size = sz;
4270
4271 errno = 0;
4274 {
4276
4278
4279 /* if write didn't set errno, assume problem is no disk space */
4282 (errcode_for_file_access(),
4284 txn->xid)));
4285 }
4286 pgstat_report_wait_end();
4287
4288 /*
4289 * Keep the transaction's final_lsn up to date with each change we send to
4290 * disk, so that ReorderBufferRestoreCleanup works correctly. (We used to
4291 * only do this on commit and abort records, but that doesn't work if a
4292 * system crash leaves a transaction without its abort record).
4293 *
4294 * Make sure not to move it backwards.
4295 */
4298
4300}
4301
4302/* Returns true, if the output plugin supports streaming, false, otherwise. */
4303static inline bool
4305{
4307
4309}
4310
4311/* Returns true, if the streaming can be started now, false, otherwise. */
4312static inline bool
4314{
4317
4318 /* We can't start streaming unless a consistent state is reached. */
4320 return false;
4321
4322 /*
4323 * We can't start streaming immediately even if the streaming is enabled
4324 * because we previously decoded this transaction and now just are
4325 * restarting.
4326 */
4329 return true;
4330
4331 return false;
4332}
4333
4334/*
4335 * Send data of a large transaction (and its subtransactions) to the
4336 * output plugin, but using the stream API.
4337 */
4338static void
4340{
4341 Snapshot snapshot_now;
4342 CommandId command_id;
4343 Size stream_bytes;
4345
4346 /* We can never reach here for a subtransaction. */
4348
4349 /*
4350 * We can't make any assumptions about base snapshot here, similar to what
4351 * ReorderBufferCommit() does. That relies on base_snapshot getting
4352 * transferred from subxact in ReorderBufferCommitChild(), but that was
4353 * not yet called as the transaction is in-progress.
4354 *
4355 * So just walk the subxacts and use the same logic here. But we only need
4356 * to do that once, when the transaction is streamed for the first time.
4357 * After that we need to reuse the snapshot from the previous run.
4358 *
4359 * Unlike DecodeCommit which adds xids of all the subtransactions in
4360 * snapshot's xip array via SnapBuildCommitTxn, we can't do that here but
4361 * we do add them to subxip array instead via ReorderBufferCopySnap. This
4362 * allows the catalog changes made in subtransactions decoded till now to
4363 * be visible.
4364 */
4366 {
4368
4369 /* make sure this transaction is streamed for the first time */
4371
4372 /* at the beginning we should have invalid command ID */
4374
4376 {
4378
4381 }
4382
4383 /*
4384 * If this transaction has no snapshot, it didn't make any changes to
4385 * the database till now, so there's nothing to decode.
4386 */
4388 {
4390 return;
4391 }
4392
4393 command_id = FirstCommandId;
4395 txn, command_id);
4396 }
4397 else
4398 {
4399 /* the transaction must have been already streamed */
4401
4402 /*
4403 * Nah, we already have snapshot from the previous streaming run. We
4404 * assume new subxacts can't move the LSN backwards, and so can't beat
4405 * the LSN condition in the previous branch (so no need to walk
4406 * through subxacts again). In fact, we must not do that as we may be
4407 * using snapshot half-way through the subxact.
4408 */
4409 command_id = txn->command_id;
4410
4411 /*
4412 * We can't use txn->snapshot_now directly because after the last
4413 * streaming run, we might have got some new sub-transactions. So we
4414 * need to add them to the snapshot.
4415 */
4417 txn, command_id);
4418
4419 /* Free the previously copied snapshot. */
4423 }
4424
4425 /*
4426 * Remember this information to be used later to update stats. We can't
4427 * update the stats here as an error while processing the changes would
4428 * lead to the accumulation of stats even though we haven't streamed all
4429 * the changes.
4430 */
4432 stream_bytes = txn->total_size;
4433
4434 /* Process and send the changes to output plugin. */
4436 command_id, true);
4437
4438 rb->streamCount += 1;
4439 rb->streamBytes += stream_bytes;
4440
4441 /* Don't consider already streamed transaction. */
4443
4444 /* update the decoding stats */
4446
4450}
4451
4452/*
4453 * Size of a change in memory.
4454 */
4457{
4459
4461 {
4462 /* fall through these, they're all similar enough */
4467 {
4469 newtup;
4472
4475
4477 {
4481 }
4482
4484 {
4488 }
4489
4490 break;
4491 }
4493 {
4495
4498
4499 break;
4500 }
4502 {
4505 break;
4506 }
4508 {
4510
4512
4516
4517 break;
4518 }
4520 {
4522
4523 break;
4524 }
4529 /* ReorderBufferChange contains everything important */
4530 break;
4531 }
4532
4534}
4535
4536
4537/*
4538 * Restore a number of changes spilled to disk back into memory.
4539 */
4543{
4548
4551
4552 /* free current entries, so we have memory for more */
4554 {
4557
4560 }
4561 txn->nentries_mem = 0;
4563
4565
4567 {
4569 ReorderBufferDiskChange *ondisk;
4570
4571 CHECK_FOR_INTERRUPTS();
4572
4574 {
4576
4577 /* first time in */
4578 if (*segno == 0)
4580
4582
4583 /*
4584 * No need to care about TLIs here, only used during a single run,
4585 * so each LSN only maps to a specific WAL record.
4586 */
4588 *segno);
4589
4591
4592 /* No harm in resetting the offset even in case of failure */
4593 file->curOffset = 0;
4594
4596 {
4597 *fd = -1;
4598 (*segno)++;
4599 continue;
4600 }
4603 (errcode_for_file_access(),
4605 path)));
4606 }
4607
4608 /*
4609 * Read the statically sized part of a change which has information
4610 * about the total size. If we couldn't read a record, we're at the
4611 * end of this file.
4612 */
4617
4618 /* eof */
4620 {
4622 *fd = -1;
4623 (*segno)++;
4624 continue;
4625 }
4628 (errcode_for_file_access(),
4632 (errcode_for_file_access(),
4634 readBytes,
4636
4638
4640
4644
4648 file->curOffset,
4650
4653 (errcode_for_file_access(),
4657 (errcode_for_file_access(),
4659 readBytes,
4661
4663
4664 /*
4665 * ok, read a full change from disk, now restore it into proper
4666 * in-memory format
4667 */
4669 restored++;
4670 }
4671
4673}
4674
4675/*
4676 * Convert change from its on-disk format to in-memory format and queue it onto
4677 * the TXN's ->changes list.
4678 *
4679 * Note: although "data" is declared char*, at entry it points to a
4680 * maxalign'd buffer, making it safe in most of this function to assume
4681 * that the pointed-to data is suitably aligned for direct access.
4682 */
4683static void
4686{
4687 ReorderBufferDiskChange *ondisk;
4688 ReorderBufferChange *change;
4689
4691
4693
4694 /* copy static part */
4696
4698
4699 /* restore individual stuff */
4701 {
4702 /* fall through these, they're all similar enough */
4708 {
4710
4713
4714 /* restore ->tuple */
4718
4719 /* reset t_data pointer into the new tuplebuf */
4722
4723 /* restore tuple data itself */
4726 }
4727
4729 {
4730 /* here, data might not be suitably aligned! */
4732
4735
4738
4739 /* restore ->tuple */
4743
4744 /* reset t_data pointer into the new tuplebuf */
4747
4748 /* restore tuple data itself */
4751 }
4752
4753 break;
4755 {
4756 Size prefix_size;
4757
4758 /* read prefix */
4762 prefix_size);
4765 data += prefix_size;
4766
4767 /* read the message */
4775
4776 break;
4777 }
4779 {
4782
4785
4786 /* read the message */
4788
4789 break;
4790 }
4792 {
4795 Size size;
4796
4798
4802
4804
4806
4812 break;
4813 }
4814 /* the base struct contains all the data, easy peasy */
4816 {
4817 Oid *relids;
4818
4822
4823 break;
4824 }
4829 break;
4830 }
4831
4833 txn->nentries_mem++;
4834
4835 /*
4836 * Update memory accounting for the restored change. We need to do this
4837 * although we don't check the memory limit when restoring the changes in
4838 * this branch (we only do that when initially queueing the changes after
4839 * decoding), because we will release the changes later, and that will
4840 * update the accounting too (subtracting the size from the counters). And
4841 * we don't want to underflow there.
4842 */
4844 ReorderBufferChangeSize(change));
4845}
4846
4847/*
4848 * Remove all on-disk stored for the passed in transaction.
4849 */
4850static void
4852{
4853 XLogSegNo first;
4855 XLogSegNo last;
4856
4859
4862
4863 /* iterate over all possible filenames, and delete them */
4865 {
4867
4871 (errcode_for_file_access(),
4873 }
4874}
4875
4876/*
4877 * Remove any leftover serialized reorder buffers from a slot directory after a
4878 * prior crash or decoding session exit.
4879 */
4880static void
4882{
4887
4889
4890 /* we're only handling directories here, skip if it's not ours */
4892 return;
4893
4896 {
4897 /* only look at names that can be ours */
4899 {
4902 spill_de->d_name);
4903
4906 (errcode_for_file_access(),
4908 path, PG_REPLSLOT_DIR, slotname)));
4909 }
4910 }
4912}
4913
4914/*
4915 * Given a replication slot, transaction ID and segment number, fill in the
4916 * corresponding spill file into 'path', which is a caller-owned buffer of size
4917 * at least MAXPGPATH.
4918 */
4919static void
4921 XLogSegNo segno)
4922{
4924
4926
4928 PG_REPLSLOT_DIR,
4931}
4932
4933/*
4934 * Delete all data spilled to disk after we've restarted/crashed. It will be
4935 * recreated when the respective slots are reused.
4936 */
4937void
4939{
4942
4945 {
4948 continue;
4949
4950 /* if it cannot be a slot, skip the directory */
4952 continue;
4953
4954 /*
4955 * ok, has to be a surviving logical slot, iterate and delete
4956 * everything starting with xid-*
4957 */
4959 }
4961}
4962
4963/* ---------------------------------------
4964 * toast reassembly support
4965 * ---------------------------------------
4966 */
4967
4968/*
4969 * Initialize per tuple toast reconstruction support.
4970 */
4971static void
4973{
4975
4977
4983}
4984
4985/*
4986 * Per toast-chunk handling for toast reconstruction
4987 *
4988 * Appends a toast chunk so we can reconstruct it when the tuple "owning" the
4989 * toasted Datum comes along.
4990 */
4991static void
4994{
4997 bool found;
4999 bool isnull;
5000 Pointer chunk;
5002 Oid chunk_id;
5004
5007
5009
5012 Assert(!isnull);
5014 Assert(!isnull);
5015
5018
5019 if (!found)
5020 {
5022 ent->num_chunks = 0;
5023 ent->last_chunk_seq = 0;
5024 ent->size = 0;
5027
5030 chunk_seq, chunk_id);
5031 }
5035
5037 Assert(!isnull);
5038
5039 /* calculate size so we can allocate the right size at once later */
5043 /* could happen due to heap_form_tuple doing its thing */
5045 else
5047
5050 ent->num_chunks++;
5052}
5053
5054/*
5055 * Rejigger change->newtuple to point to in-memory toast tuples instead of
5056 * on-disk toast tuples that may no longer exist (think DROP TABLE or VACUUM).
5057 *
5058 * We cannot replace unchanged toast tuples though, so those will still point
5059 * to on-disk toast data.
5060 *
5061 * While updating the existing change with detoasted tuple data, we need to
5062 * update the memory accounting info, because the change size will differ.
5063 * Otherwise the accounting may get out of sync, triggering serialization
5064 * at unexpected times.
5065 *
5066 * We simply subtract size of the change before rejiggering the tuple, and
5067 * then add the new size. This makes it look like the change was removed
5068 * and then added back, except it only tweaks the accounting info.
5069 *
5070 * In particular it can't trigger serialization, which would be pointless
5071 * anyway as it happens during commit processing right before handing
5072 * the change to the output plugin.
5073 */
5074static void
5077{
5078 TupleDesc desc;
5081 bool *isnull;
5084 Relation toast_rel;
5086 MemoryContext oldcontext;
5089
5090 /* no toast tuples changed */
5092 return;
5093
5094 /*
5095 * We're going to modify the size of the change. So, to make sure the
5096 * accounting is correct we record the current change size and then after
5097 * re-computing the change we'll subtract the recorded size and then
5098 * re-add the new change size at the end. We don't immediately subtract
5099 * the old size because if there is any error before we add the new size,
5100 * we will release the changes and that will update the accounting info
5101 * (subtracting the size from the counters). And we don't want to
5102 * underflow there.
5103 */
5105
5107
5108 /* we should only have toast tuples in an INSERT or UPDATE */
5110
5111 desc = RelationGetDescr(relation);
5112
5117
5119
5120 /* should we allocate from stack instead? */
5124
5126
5128
5130 {
5134
5135 /* va_rawsize is the size of the original datum -- including header */
5136 varatt_external toast_pointer;
5139 varlena *reconstructed;
5142
5144 continue;
5145
5146 /* not a varlena datatype */
5148 continue;
5149
5150 /* no data */
5152 continue;
5153
5154 /* ok, we know we have a toast datum */
5156
5157 /* no need to do anything if the tuple isn't external */
5159 continue;
5160
5162
5163 /*
5164 * Check whether the toast tuple changed, replace if so.
5165 */
5168 &toast_pointer.va_valueid,
5169 HASH_FIND,
5170 NULL);
5172 continue;
5173
5174 new_datum =
5176
5178
5180
5181 ent->reconstructed = reconstructed;
5182
5183 /* stitch toast tuple back together from its parts */
5185 {
5189 Pointer chunk;
5190
5194
5198
5200 VARDATA(chunk),
5203 }
5205
5206 /* make sure its marked as compressed or not */
5209 else
5211
5213 redirect_pointer.pointer = reconstructed;
5214
5218
5220 }
5221
5222 /*
5223 * Build tuple in separate memory & copy tuple back into the tuplebuf
5224 * passed to the output plugin. We can't directly heap_fill_tuple() into
5225 * the tuplebuf because attrs[] will point back into the current content.
5226 */
5230
5233
5234 /*
5235 * free resources we won't further need, more persistent stuff will be
5236 * free'd in ReorderBufferToastReset().
5237 */
5238 RelationClose(toast_rel);
5241 {
5244 }
5247 pfree(isnull);
5248
5249 MemoryContextSwitchTo(oldcontext);
5250
5251 /* subtract the old change size */
5253 /* now add the change back, with the correct size */
5255 ReorderBufferChangeSize(change));
5256}
5257
5258/*
5259 * Free all resources allocated for toast reconstruction.
5260 */
5261static void
5263{
5266
5268 return;
5269
5270 /* sequentially walk over the hash and free everything */
5273 {
5275
5278
5280 {
5281 ReorderBufferChange *change =
5283
5286 }
5287 }
5288
5291}
5292
5293
5294/* ---------------------------------------
5295 * Visibility support for logical decoding
5296 *
5297 *
5298 * Lookup actual cmin/cmax values when using decoding snapshot. We can't
5299 * always rely on stored cmin/cmax values because of two scenarios:
5300 *
5301 * * A tuple got changed multiple times during a single transaction and thus
5302 * has got a combo CID. Combo CIDs are only valid for the duration of a
5303 * single transaction.
5304 * * A tuple with a cmin but no cmax (and thus no combo CID) got
5305 * deleted/updated in another transaction than the one which created it
5306 * which we are looking at right now. As only one of cmin, cmax or combo CID
5307 * is actually stored in the heap we don't have access to the value we
5308 * need anymore.
5309 *
5310 * To resolve those problems we have a per-transaction hash of (cmin,
5311 * cmax) tuples keyed by (relfilelocator, ctid) which contains the actual
5312 * (cmin, cmax) values. That also takes care of combo CIDs by simply
5313 * not caring about them at all. As we have the real cmin/cmax values
5314 * combo CIDs aren't interesting.
5315 *
5316 * As we only care about catalog tuples here the overhead of this
5317 * hashtable should be acceptable.
5318 *
5319 * Heap rewrites complicate this a bit, check rewriteheap.c for
5320 * details.
5321 * -------------------------------------------------------------------------
5322 */
5323
5324/* struct for sorting mapping files by LSN efficiently */
5326{
5329} RewriteMappingFile;
5330
5331#ifdef NOT_USED
5332static void
5334{
5337
5340 {
5342 ent->key.rlocator.dbOid,
5343 ent->key.rlocator.spcOid,
5344 ent->key.rlocator.relNumber,
5347 ent->cmin,
5348 ent->cmax
5349 );
5350 }
5351}
5352#endif
5353
5354/*
5355 * Apply a single mapping file to tuplecid_data.
5356 *
5357 * The mapping file has to have been verified to be a) committed b) for our
5358 * transaction c) applied in LSN order.
5359 */
5360static void
5362{
5366 LogicalRewriteMappingData map;
5367
5372 (errcode_for_file_access(),
5374
5375 while (true)
5376 {
5380 bool found;
5381
5382 /* be careful about padding */
5384
5385 /* read all mappings till the end of the file */
5388 pgstat_report_wait_end();
5389
5392 (errcode_for_file_access(),
5394 path)));
5396 break;
5399 (errcode_for_file_access(),
5401 path, readBytes,
5403
5406 &key.tid);
5407
5408
5411
5412 /* no existing mapping, no need to update */
5414 continue;
5415
5418 &key.tid);
5419
5422
5423 if (found)
5424 {
5425 /*
5426 * Make sure the existing mapping makes sense. We sometime update
5427 * old records that did not yet have a cmax (e.g. pg_class' own
5428 * entry while rewriting it) during rewrites, so allow that.
5429 */
5432 }
5433 else
5434 {
5435 /* update mapping */
5439 }
5440 }
5441
5444 (errcode_for_file_access(),
5446}
5447
5448
5449/*
5450 * Check whether the TransactionId 'xid' is in the pre-sorted array 'xip'.
5451 */
5452static bool
5454{
5457}
5458
5459/*
5460 * list_sort() comparator for sorting RewriteMappingFiles in LSN order.
5461 */
5462static int
5464{
5467
5469}
5470
5471/*
5472 * Apply any existing logical remapping files if there are any targeted at our
5473 * transaction for relid.
5474 */
5475static void
5477{
5481 ListCell *file;
5483
5486 {
5493 f_lo;
5494 RewriteMappingFile *f;
5495
5498 continue;
5499
5500 /* Ignore files that aren't ours */
5502 continue;
5503
5508
5510
5511 /* mapping for another database */
5513 continue;
5514
5515 /* mapping for another relation */
5517 continue;
5518
5519 /* did the creating transaction abort? */
5521 continue;
5522
5523 /* not for our transaction */
5525 continue;
5526
5527 /* ok, relevant, queue for apply */
5531 files = lappend(files, f);
5532 }
5534
5535 /* sort files so we apply them in LSN order */
5537
5538 foreach(file, files)
5539 {
5541
5543 snapshot->subxip[0]);
5545 pfree(f);
5546 }
5547}
5548
5549/*
5550 * Lookup cmin/cmax of a tuple, during logical decoding where we can't rely on
5551 * combo CIDs.
5552 */
5553bool
5555 Snapshot snapshot,
5558{
5561 ForkNumber forkno;
5562 BlockNumber blockno;
5564
5565 /*
5566 * Return unresolved if tuplecid_data is not valid. That's because when
5567 * streaming in-progress transactions we may run into tuples with the CID
5568 * before actually decoding them. Think e.g. about INSERT followed by
5569 * TRUNCATE, where the TRUNCATE may not be decoded yet when applying the
5570 * INSERT. So in such cases, we assume the CID is from the future
5571 * command.
5572 */
5574 return false;
5575
5576 /* be careful about padding */
5578
5580
5581 /*
5582 * get relfilelocator from the buffer, no convenient way to access it
5583 * other than that.
5584 */
5586
5587 /* tuples can only be in the main fork */
5590
5592 &key.tid);
5593
5594restart:
5597
5598 /*
5599 * failed to find a mapping, check whether the table was rewritten and
5600 * apply mapping if so, but only do that once - there can be no new
5601 * mappings while we are in here since we have to hold a lock on the
5602 * relation.
5603 */
5605 {
5607 /* now check but don't update for a mapping again */
5609 goto restart;
5610 }
5612 return false;
5613
5614 if (cmin)
5615 *cmin = ent->cmin;
5616 if (cmax)
5617 *cmax = ent->cmax;
5618 return true;
5619}
5620
5621/*
5622 * Count invalidation messages of specified transaction.
5623 *
5624 * Returns number of messages, and msgs is set to the pointer of the linked
5625 * list for the messages.
5626 */
5627uint32
5629 SharedInvalidationMessage **msgs)
5630{
5631 ReorderBufferTXN *txn;
5632
5634 false);
5635
5637 return 0;
5638
5639 *msgs = txn->invalidations;
5640
5642}
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:4378
#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:952
HTAB * hash_create(const char *tabname, int64 nelem, const HASHCTL *info, int flags)
Definition dynahash.c:358
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:1037
void heap_deform_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *values, bool *isnull)
Definition heaptuple.c:1266
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:1266
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:124
bool TransactionIdIsInProgress(TransactionId xid)
Definition procarray.c:1401
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:5464
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:5076
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Definition reorderbuffer.c:3651
static void ReorderBufferStreamCommit(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:1984
void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)
Definition reorderbuffer.c:3355
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:2821
static void ReorderBufferAccumulateInvalidations(SharedInvalidationMessage **invals_out, uint32 *ninvals_out, SharedInvalidationMessage *msgs_new, Size nmsgs_new)
Definition reorderbuffer.c:3503
static ReorderBufferTXN * ReorderBufferLargestTXN(ReorderBuffer *rb)
Definition reorderbuffer.c:3803
void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
Definition reorderbuffer.c:3454
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
Definition reorderbuffer.c:3324
static void ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:4852
static void ReorderBufferToastInitHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:4973
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, TimestampTz abort_time)
Definition reorderbuffer.c:3085
static bool ReorderBufferCanStartStreaming(ReorderBuffer *rb)
Definition reorderbuffer.c:4314
static void ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, Snapshot snapshot_now, CommandId command_id, XLogRecPtr last_lsn, ReorderBufferChange *specinsert)
Definition reorderbuffer.c:2166
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
Definition reorderbuffer.c:3723
void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Definition reorderbuffer.c:3220
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:5555
static void ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition reorderbuffer.c:4993
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:2958
uint32 ReorderBufferGetInvalidations(ReorderBuffer *rb, TransactionId xid, SharedInvalidationMessage **msgs)
Definition reorderbuffer.c:5629
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Definition reorderbuffer.c:3178
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:3689
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:4457
static void AssertChangeLsnOrder(ReorderBufferTXN *txn)
Definition reorderbuffer.c:1014
static bool ReorderBufferCanStream(ReorderBuffer *rb)
Definition reorderbuffer.c:4305
static int ReorderBufferTXNSizeCompare(const pairingheap_node *a, const pairingheap_node *b, void *arg)
Definition reorderbuffer.c:3787
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:2937
bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid, XLogRecPtr prepare_lsn, XLogRecPtr end_lsn, TimestampTz prepare_time, ReplOriginId origin_id, XLogRecPtr origin_lsn)
Definition reorderbuffer.c:2905
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:2999
static void ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn, int fd, ReorderBufferChange *change)
Definition reorderbuffer.c:4090
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
Definition reorderbuffer.c:3540
void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, ReplOriginId origin_id, XLogRecPtr origin_lsn)
Definition reorderbuffer.c:2882
int debug_logical_replication_streaming
Definition reorderbuffer.c:230
void ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
Definition reorderbuffer.c:3581
static void ReorderBufferSerializeReserve(ReorderBuffer *rb, Size sz)
Definition reorderbuffer.c:3770
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:3740
static void ReorderBufferExecuteInvalidations(uint32 nmsgs, SharedInvalidationMessage *msgs)
Definition reorderbuffer.c:3639
static void ReorderBufferIterTXNFinish(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
Definition reorderbuffer.c:1505
void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
Definition reorderbuffer.c:3306
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:3844
static bool ReorderBufferCheckAndTruncateAbortedTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:1775
static void ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn, char *data)
Definition reorderbuffer.c:4685
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:4882
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:5454
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:5263
static void ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:3995
static void UpdateLogicalMappings(HTAB *tuplecid_data, Oid relid, Snapshot snapshot)
Definition reorderbuffer.c:5477
static void ReorderBufferQueueInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
Definition reorderbuffer.c:3481
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:3251
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:3894
static void ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb, ReorderBufferChange *change, ReorderBufferTXN *txn, bool addition, Size sz)
Definition reorderbuffer.c:3383
static void ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
Definition reorderbuffer.c:4340
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Definition reorderbuffer.c:3293
static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn, TXNEntryFile *file, XLogSegNo *segno)
Definition reorderbuffer.c:4542
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:4921
static void ApplyLogicalMappingFile(HTAB *tuplecid_data, const char *fname)
Definition reorderbuffer.c:5362
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
Definition reorderbuffer.c:3131
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:2212
@ 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:268
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition snapbuild.c:307
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition snapbuild.c:280
void SetupHistoricSnapshot(Snapshot historic_snapshot, HTAB *tuplecids)
Definition snapmgr.c:1669
Definition tupdesc.h:69
Definition dirent.c:26
Definition elog.h:420
Definition hsearch.h:66
Definition hsearch.h:121
Definition dynahash.c:222
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::@117::@119 truncate
bool clear_toast_afterwards
Definition reorderbuffer.h:101
struct ReorderBufferChange::@117::@121 tuplecid
union ReorderBufferChange::@117 data
struct ReorderBufferChange::@117::@122 inval
struct ReorderBufferChange::@117::@120 msg
SharedInvalidationMessage * invalidations
Definition reorderbuffer.h:155
struct ReorderBufferChange::@117::@118 tp
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:255
Definition regguts.h:323
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:193
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:69
void RollbackAndReleaseCurrentSubTransaction(void)
Definition xact.c:4819
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
Definition xlog_internal.h:101
◆ 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 5362 of file reorderbuffer.c.
5363{
5367 LogicalRewriteMappingData map;
5368
5373 (errcode_for_file_access(),
5375
5376 while (true)
5377 {
5381 bool found;
5382
5383 /* be careful about padding */
5385
5386 /* read all mappings till the end of the file */
5389 pgstat_report_wait_end();
5390
5393 (errcode_for_file_access(),
5395 path)));
5397 break;
5400 (errcode_for_file_access(),
5402 path, readBytes,
5404
5407 &key.tid);
5408
5409
5412
5413 /* no existing mapping, no need to update */
5415 continue;
5416
5419 &key.tid);
5420
5423
5424 if (found)
5425 {
5426 /*
5427 * Make sure the existing mapping makes sense. We sometime update
5428 * old records that did not yet have a cmax (e.g. pg_class' own
5429 * entry while rewriting it) during rewrites, so allow that.
5430 */
5433 }
5434 else
5435 {
5436 /* update mapping */
5440 }
5441 }
5442
5445 (errcode_for_file_access(),
5447}
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 5464 of file reorderbuffer.c.
5465{
5468
5470}
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 3085 of file reorderbuffer.c.
3087{
3088 ReorderBufferTXN *txn;
3089
3091 false);
3092
3093 /* unknown, nothing to remove */
3095 return;
3096
3097 txn->abort_time = abort_time;
3098
3099 /* For streamed transactions notify the remote node about the abort. */
3101 {
3103
3104 /*
3105 * We might have decoded changes for this transaction that could load
3106 * the cache as per the current transaction's view (consider DDL's
3107 * happened in this transaction). We don't want the decoding of future
3108 * transactions to use those cache entries so execute only the inval
3109 * messages in this transaction.
3110 */
3113 txn->invalidations);
3114 }
3115
3116 /* cosmetic... */
3117 txn->final_lsn = lsn;
3118
3119 /* remove potential on-disk data, and deallocate */
3121}
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 3131 of file reorderbuffer.c.
3132{
3134
3135 /*
3136 * Iterate through all (potential) toplevel TXNs and abort all that are
3137 * older than what possibly can be running. Once we've found the first
3138 * that is alive we stop, there might be some that acquired an xid earlier
3139 * but started writing later, but it's unlikely and they will be cleaned
3140 * up in a later call to this function.
3141 */
3143 {
3144 ReorderBufferTXN *txn;
3145
3147
3149 {
3151
3152 /* Notify the remote node about the crash/immediate restart. */
3155
3156 /* remove potential on-disk data, and deallocate this tx */
3158 }
3159 else
3160 return;
3161 }
3162}
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 3503 of file reorderbuffer.c.
3507{
3509 {
3513 }
3514 else
3515 {
3516 /* Enlarge the array of inval messages */
3517 *invals_out =
3523 }
3524}
References fb(), 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 3581 of file reorderbuffer.c.
3584{
3585 ReorderBufferTXN *txn;
3586 MemoryContext oldcontext;
3587
3589
3591
3592 /*
3593 * Collect all the invalidations under the top transaction, if available,
3594 * so that we can execute them all together. See comments
3595 * ReorderBufferAddInvalidations.
3596 */
3597 txn = rbtxn_get_toptxn(txn);
3598
3599 Assert(nmsgs > 0);
3600
3602 {
3603 /*
3604 * Check the transaction has enough space for storing distributed
3605 * invalidation messages.
3606 */
3608 {
3609 /*
3610 * Mark the invalidation message as overflowed and free up the
3611 * messages accumulated so far.
3612 */
3614
3616 {
3619 txn->ninvalidations_distributed = 0;
3620 }
3621 }
3622 else
3624 &txn->ninvalidations_distributed,
3625 msgs, nmsgs);
3626 }
3627
3628 /* Queue the invalidation messages into the transaction */
3630
3631 MemoryContextSwitchTo(oldcontext);
3632}
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 3540 of file reorderbuffer.c.
3543{
3544 ReorderBufferTXN *txn;
3545 MemoryContext oldcontext;
3546
3548
3550
3551 /*
3552 * Collect all the invalidations under the top transaction, if available,
3553 * so that we can execute them all together. See comments atop this
3554 * function.
3555 */
3556 txn = rbtxn_get_toptxn(txn);
3557
3558 Assert(nmsgs > 0);
3559
3561 &txn->ninvalidations,
3562 msgs, nmsgs);
3563
3565
3566 MemoryContextSwitchTo(oldcontext);
3567}
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 3355 of file reorderbuffer.c.
3357{
3359
3362
3364}
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 3454 of file reorderbuffer.c.
3458{
3460 ReorderBufferTXN *txn;
3461
3463
3469 change->lsn = lsn;
3470 change->txn = txn;
3472
3474 txn->ntuplecids++;
3475}
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 3306 of file reorderbuffer.c.
3308{
3310
3313
3315}
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 4314 of file reorderbuffer.c.
4315{
4318
4319 /* We can't start streaming unless a consistent state is reached. */
4321 return false;
4322
4323 /*
4324 * We can't start streaming immediately even if the streaming is enabled
4325 * because we previously decoded this transaction and now just are
4326 * restarting.
4327 */
4330 return true;
4331
4332 return false;
4333}
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 4305 of file reorderbuffer.c.
References fb(), and LogicalDecodingContext::streaming.
Referenced by ReorderBufferCanStartStreaming(), and ReorderBufferProcessPartialChange().
◆ ReorderBufferChangeMemoryUpdate()
|
static |
Definition at line 3383 of file reorderbuffer.c.
3387{
3388 ReorderBufferTXN *toptxn;
3389
3390 Assert(txn || change);
3391
3392 /*
3393 * Ignore tuple CID changes, because those are not evicted when reaching
3394 * memory limit. So we just don't count them, because it might easily
3395 * trigger a pointless attempt to spill.
3396 */
3398 return;
3399
3401 return;
3402
3404 txn = change->txn;
3406
3407 /*
3408 * Update the total size in top level as well. This is later used to
3409 * compute the decoding stats.
3410 */
3411 toptxn = rbtxn_get_toptxn(txn);
3412
3414 {
3416
3419
3420 /* Update the total size in the top transaction. */
3422
3423 /* Update the max-heap */
3427 }
3428 else
3429 {
3433
3434 /* Update the total size in the top transaction. */
3436
3437 /* Update the max-heap */
3441 }
3442
3444}
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 4457 of file reorderbuffer.c.
4458{
4460
4462 {
4463 /* fall through these, they're all similar enough */
4468 {
4470 newtup;
4473
4476
4478 {
4482 }
4483
4485 {
4489 }
4490
4491 break;
4492 }
4494 {
4496
4499
4500 break;
4501 }
4503 {
4506 break;
4507 }
4509 {
4511
4513
4517
4518 break;
4519 }
4521 {
4523
4524 break;
4525 }
4530 /* ReorderBufferChange contains everything important */
4531 break;
4532 }
4533
4535}
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 3894 of file reorderbuffer.c.
3895{
3896 ReorderBufferTXN *txn;
3898
3900 {
3901 /*
3902 * Update the statistics as the memory usage has reached the limit. We
3903 * report the statistics update later in this function since we can
3904 * update the slot statistics altogether while streaming or
3905 * serializing transactions in most cases.
3906 */
3907 rb->memExceededCount += 1;
3908 }
3910 {
3911 /*
3912 * Bail out if debug_logical_replication_streaming is buffered and we
3913 * haven't exceeded the memory limit.
3914 */
3915 return;
3916 }
3917
3918 /*
3919 * If debug_logical_replication_streaming is immediate, loop until there's
3920 * no change. Otherwise, loop until we reach under the memory limit. One
3921 * might think that just by evicting the largest (sub)transaction we will
3922 * come under the memory limit based on assumption that the selected
3923 * transaction is at least as large as the most recent change (which
3924 * caused us to go over the memory limit). However, that is not true
3925 * because a user can reduce the logical_decoding_work_mem to a smaller
3926 * value before the most recent change.
3927 */
3930 rb->size > 0))
3931 {
3932 /*
3933 * Pick the largest non-aborted transaction and evict it from memory
3934 * by streaming, if possible. Otherwise, spill to disk.
3935 */
3938 {
3939 /* we know there has to be one, because the size is not zero */
3943
3944 /* skip the transaction if aborted */
3946 continue;
3947
3949 }
3950 else
3951 {
3952 /*
3953 * Pick the largest transaction (or subtransaction) and evict it
3954 * from memory by serializing it to disk.
3955 */
3957
3958 /* we know there has to be one, because the size is not zero */
3959 Assert(txn);
3962
3963 /* skip the transaction if aborted */
3965 continue;
3966
3968 }
3969
3970 /*
3971 * After eviction, the transaction should have no entries in memory,
3972 * and should use 0 bytes for changes.
3973 */
3976
3977 /*
3978 * We've reported the memExceededCount update while streaming or
3979 * serializing the transaction.
3980 */
3982 }
3983
3986
3987 /* We must be under the memory limit now. */
3989}
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 4882 of file reorderbuffer.c.
4883{
4888
4890
4891 /* we're only handling directories here, skip if it's not ours */
4893 return;
4894
4897 {
4898 /* only look at names that can be ours */
4900 {
4903 spill_de->d_name);
4904
4907 (errcode_for_file_access(),
4909 path, PG_REPLSLOT_DIR, slotname)));
4910 }
4911 }
4913}
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 2882 of file reorderbuffer.c.
2886{
2887 ReorderBufferTXN *txn;
2888
2890 false);
2891
2892 /* unknown transaction, nothing to replay */
2894 return;
2895
2897 origin_id, origin_lsn);
2898}
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, MemoryContextAllocZero(), ReorderBufferTXN::nsubtxns, qsort, ReorderBufferTXN::subtxns, ReorderBufferTXN::xid, and xidComparator().
Referenced by ReorderBufferProcessTXN(), ReorderBufferSaveTXNSnapshot(), and ReorderBufferStreamTXN().
◆ ReorderBufferExecuteInvalidations()
|
static |
Definition at line 3639 of file reorderbuffer.c.
3640{
3642
3645}
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 2999 of file reorderbuffer.c.
3004{
3005 ReorderBufferTXN *txn;
3006 XLogRecPtr prepare_end_lsn;
3007 TimestampTz prepare_time;
3008
3010
3011 /* unknown transaction, nothing to do */
3013 return;
3014
3015 /*
3016 * By this time the txn has the prepare record information, remember it to
3017 * be later used for rollback.
3018 */
3019 prepare_end_lsn = txn->end_lsn;
3020 prepare_time = txn->prepare_time;
3021
3022 /* add the gid in the txn */
3024
3025 /*
3026 * It is possible that this transaction is not decoded at prepare time
3027 * either because by that time we didn't have a consistent snapshot, or
3028 * two_phase was not enabled, or it was decoded earlier but we have
3029 * restarted. We only need to send the prepare if it was not decoded
3030 * earlier. We don't need to decode the xact for aborts if it is not done
3031 * already.
3032 */
3034 {
3035 /*
3036 * txn must have been marked as a prepared transaction and skipped but
3037 * not sent a prepare. Also, the prepare info must have been updated
3038 * in txn even if we skip prepare.
3039 */
3043
3044 /*
3045 * By this time the txn has the prepare record information and it is
3046 * important to use that so that downstream gets the accurate
3047 * information. If instead, we have passed commit information here
3048 * then downstream can behave as it has already replayed commit
3049 * prepared after the restart.
3050 */
3053 }
3054
3055 txn->final_lsn = commit_lsn;
3056 txn->end_lsn = end_lsn;
3057 txn->commit_time = commit_time;
3058 txn->origin_id = origin_id;
3059 txn->origin_lsn = origin_lsn;
3060
3063 else
3065
3066 /* cleanup: make sure there's no cache pollution */
3068 txn->invalidations);
3070}
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 3178 of file reorderbuffer.c.
3179{
3180 ReorderBufferTXN *txn;
3181
3183 false);
3184
3185 /* unknown, nothing to forget */
3187 return;
3188
3189 /* this transaction mustn't be streamed */
3191
3192 /* cosmetic... */
3193 txn->final_lsn = lsn;
3194
3195 /*
3196 * Process only cache invalidation messages in this transaction if there
3197 * are any. Even if we're not interested in the transaction's contents, it
3198 * could have manipulated the catalog and we need to update the caches
3199 * according to that.
3200 */
3203 txn->invalidations);
3204 else
3206
3207 /* remove potential on-disk data, and deallocate */
3209}
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(), ReorderBufferResetTXN(), 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 3689 of file reorderbuffer.c.
3690{
3691 dlist_iter iter;
3693 size_t xcnt = 0;
3694
3695 /* Quick return if the list is empty */
3698
3699 /* Initialize XID array */
3702 {
3704 catchange_node,
3705 iter.cur);
3706
3708
3709 xids[xcnt++] = txn->xid;
3710 }
3711
3713
3715 return xids;
3716}
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 5629 of file reorderbuffer.c.
5631{
5632 ReorderBufferTXN *txn;
5633
5635 false);
5636
5638 return 0;
5639
5640 *msgs = txn->invalidations;
5641
5643}
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 3251 of file reorderbuffer.c.
3253{
3258
3261
3262 /*
3263 * Force invalidations to happen outside of a valid transaction - that way
3264 * entries will just be marked as invalid without accessing the catalog.
3265 * That's advantageous because we don't need to setup the full state
3266 * necessary for catalog access.
3267 */
3269 AbortCurrentTransaction();
3270
3273
3275 {
3278 CurrentResourceOwner = cowner;
3279 }
3280}
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 3220 of file reorderbuffer.c.
3221{
3222 ReorderBufferTXN *txn;
3223
3225 false);
3226
3227 /* unknown, nothing to do */
3229 return;
3230
3231 /*
3232 * Process cache invalidation messages if there are any. Even if we're not
3233 * interested in the transaction's contents, it could have manipulated the
3234 * catalog and we need to update the caches according to that.
3235 */
3238 txn->invalidations);
3239 else
3241}
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 int32 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 int32 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 3844 of file reorderbuffer.c.
3845{
3846 dlist_iter iter;
3849
3850 /* Find the largest top-level transaction having a base snapshot. */
3852 {
3853 ReorderBufferTXN *txn;
3854
3856
3857 /* must not be a subtxn */
3859 /* base_snapshot must be set */
3861
3862 /* Don't consider these kinds of transactions for eviction. */
3864 !rbtxn_has_streamable_change(txn) ||
3865 rbtxn_is_aborted(txn))
3866 continue;
3867
3868 /* Find the largest of the eviction candidates. */
3870 (txn->total_size > 0))
3871 {
3872 largest = txn;
3874 }
3875 }
3876
3878}
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 3803 of file reorderbuffer.c.
3804{
3806
3807 /* Get the largest transaction from the max-heap */
3810
3814
3816}
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 2958 of file reorderbuffer.c.
2960{
2961 ReorderBufferTXN *txn;
2962
2964 false);
2965
2966 /* unknown transaction, nothing to replay */
2968 return;
2969
2970 /*
2971 * txn must have been marked as a prepared transaction and must have
2972 * neither been skipped nor sent a prepare. Also, the prepare info must
2973 * have been updated in it by now.
2974 */
2977
2979
2982
2983 /*
2984 * Send a prepare if not already done so. This might occur if we have
2985 * detected a concurrent abort while replaying the non-streaming
2986 * transaction.
2987 */
2989 {
2992 }
2993}
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 2212 of file reorderbuffer.c.
2217{
2226
2227 /* build data to be able to lookup the CommandIds of catalog tuples */
2229
2230 /* setup the initial snapshot */
2232
2233 /*
2234 * Decoding needs access to syscaches et al., which in turn use
2235 * heavyweight locks and such. Thus we need to have enough state around to
2236 * keep track of those. The easiest way is to simply use a transaction
2237 * internally. That also allows us to easily enforce that nothing writes
2238 * to the database by checking for xid assignments.
2239 *
2240 * When we're called via the SQL SRF there's already a transaction
2241 * started, so start an explicit subtransaction there.
2242 */
2244
2245 PG_TRY();
2246 {
2247 ReorderBufferChange *change;
2249 * changes */
2250
2253 else
2254 StartTransactionCommand();
2255
2256 /*
2257 * We only need to send begin/begin-prepare for non-streamed
2258 * transactions.
2259 */
2260 if (!streaming)
2261 {
2264 else
2266 }
2267
2270 {
2272 Oid reloid;
2273
2274 CHECK_FOR_INTERRUPTS();
2275
2276 /*
2277 * We can't call start stream callback before processing first
2278 * change.
2279 */
2281 {
2282 if (streaming)
2283 {
2287 }
2288 }
2289
2290 /*
2291 * Enforce correct ordering of changes, merged from multiple
2292 * subtransactions. The changes may have the same LSN due to
2293 * MULTI_INSERT xlog records.
2294 */
2296
2298
2299 /*
2300 * Set the current xid to detect concurrent aborts. This is
2301 * required for the cases when we decode the changes before the
2302 * COMMIT record is processed.
2303 */
2305 {
2308 }
2309
2311 {
2313
2314 /*
2315 * Confirmation for speculative insertion arrived. Simply
2316 * use as a normal record. It'll be cleaned up at the end
2317 * of INSERT processing.
2318 */
2322 change = specinsert;
2324
2325 /* intentionally fall through */
2326 pg_fallthrough;
2330 Assert(snapshot_now);
2331
2334
2335 /*
2336 * Mapped catalog tuple without data, emitted while
2337 * catalog table was in the process of being rewritten. We
2338 * can fail to look up the relfilenumber, because the
2339 * relmapper has no "historic" view, in contrast to the
2340 * normal catalog during decoding. Thus repeated rewrites
2341 * can cause a lookup failure. That's OK because we do not
2342 * decode catalog changes anyway. Normally such tuples
2343 * would be skipped over below, but we can't identify
2344 * whether the table should be logically logged without
2345 * mapping the relfilenumber to the oid.
2346 */
2354 MAIN_FORKNUM).str);
2355
2356 relation = RelationIdGetRelation(reloid);
2357
2360 reloid,
2362 MAIN_FORKNUM).str);
2363
2366
2367 /*
2368 * Ignore temporary heaps created during DDL unless the
2369 * plugin has asked for them.
2370 */
2373
2374 /*
2375 * For now ignore sequence changes entirely. Most of the
2376 * time they don't log changes using records we
2377 * understand, so it doesn't make sense to handle the few
2378 * cases we do.
2379 */
2382
2383 /* user-triggered change */
2385 {
2388 streaming);
2389
2390 /*
2391 * Only clear reassembled toast chunks if we're sure
2392 * they're not required anymore. The creator of the
2393 * tuple tells us.
2394 */
2397 }
2398 /* we're not interested in toast deletions */
2400 {
2401 /*
2402 * Need to reassemble the full toasted Datum in
2403 * memory, to ensure the chunks don't get reused till
2404 * we're done remove it from the list of this
2405 * transaction's changes. Otherwise it will get
2406 * freed/reused while restoring spooled data from
2407 * disk.
2408 */
2410
2413 change);
2414 }
2415
2416 change_done:
2417
2418 /*
2419 * If speculative insertion was confirmed, the record
2420 * isn't needed anymore.
2421 */
2423 {
2426 }
2427
2429 {
2430 RelationClose(relation);
2431 relation = NULL;
2432 }
2433 break;
2434
2436
2437 /*
2438 * Speculative insertions are dealt with by delaying the
2439 * processing of the insert until the confirmation record
2440 * arrives. For that we simply unlink the record from the
2441 * chain, so it does not get freed/reused while restoring
2442 * spooled data from disk.
2443 *
2444 * This is safe in the face of concurrent catalog changes
2445 * because the relevant relation can't be changed between
2446 * speculative insertion and confirmation due to
2447 * CheckTableNotInUse() and locking.
2448 */
2449
2450 /* Previous speculative insertion must be aborted */
2452
2453 /* and memorize the pending insertion */
2455 specinsert = change;
2456 break;
2457
2459
2460 /*
2461 * Abort for speculative insertion arrived. So cleanup the
2462 * specinsert tuple and toast hash.
2463 *
2464 * Note that we get the spec abort change for each toast
2465 * entry but we need to perform the cleanup only the first
2466 * time we get it for the main table.
2467 */
2469 {
2470 /*
2471 * We must clean the toast hash before processing a
2472 * completely new tuple to avoid confusion about the
2473 * previous tuple's toast chunks.
2474 */
2477
2478 /* We don't need this record anymore. */
2481 }
2482 break;
2483
2485 {
2489 Relation *relations;
2490
2493 {
2495 Relation rel;
2496
2497 rel = RelationIdGetRelation(relid);
2498
2501
2503 continue;
2504
2505 relations[nrelations++] = rel;
2506 }
2507
2508 /* Apply the truncate. */
2510 relations, change,
2511 streaming);
2512
2515
2516 break;
2517 }
2518
2521 break;
2522
2524 /* Execute the invalidation messages locally */
2527 break;
2528
2530 /* get rid of the old */
2532
2533 if (snapshot_now->copied)
2534 {
2536 snapshot_now =
2538 txn, command_id);
2539 }
2540
2541 /*
2542 * Restored from disk, need to be careful not to double
2543 * free. We could introduce refcounting for that, but for
2544 * now this seems infrequent enough not to care.
2545 */
2547 {
2548 snapshot_now =
2550 txn, command_id);
2551 }
2552 else
2553 {
2555 }
2556
2557 /* and continue with the new one */
2559 break;
2560
2563
2565 {
2567
2568 if (!snapshot_now->copied)
2569 {
2570 /* we don't use the global one anymore */
2572 txn, command_id);
2573 }
2574
2575 snapshot_now->curcid = command_id;
2576
2579 }
2580
2581 break;
2582
2585 break;
2586 }
2587
2588 /*
2589 * It is possible that the data is not sent to downstream for a
2590 * long time either because the output plugin filtered it or there
2591 * is a DDL that generates a lot of data that is not processed by
2592 * the plugin. So, in such cases, the downstream can timeout. To
2593 * avoid that we try to send a keepalive message if required.
2594 * Trying to send a keepalive message after every change has some
2595 * overhead, but testing showed there is no noticeable overhead if
2596 * we do it after every ~100 changes.
2597 */
2598#define CHANGES_THRESHOLD 100
2599
2601 {
2603 changes_count = 0;
2604 }
2605 }
2606
2607 /* speculative insertion record must be freed by now */
2609
2610 /* clean up the iterator */
2613
2614 /*
2615 * Update total transaction count and total bytes processed by the
2616 * transaction and its subtransactions. Ensure to not count the
2617 * streamed transaction multiple times.
2618 *
2619 * Note that the statistics computation has to be done after
2620 * ReorderBufferIterTXNFinish as it releases the serialized change
2621 * which we have already accounted in ReorderBufferIterTXNNext.
2622 */
2624 rb->totalTxns++;
2625
2627
2628 /*
2629 * Done with current changes, send the last message for this set of
2630 * changes depending upon streaming mode.
2631 */
2632 if (streaming)
2633 {
2635 {
2638 }
2639 }
2640 else
2641 {
2642 /*
2643 * Call either PREPARE (for two-phase transactions) or COMMIT (for
2644 * regular ones).
2645 */
2647 {
2651 }
2652 else
2654 }
2655
2656 /* this is just a sanity check against bad output plugin behaviour */
2659 GetCurrentTransactionId());
2660
2661 /*
2662 * Remember the command ID and snapshot for the next set of changes in
2663 * streaming mode.
2664 */
2665 if (streaming)
2667 else if (snapshot_now->copied)
2669
2670 /* cleanup */
2672
2673 /*
2674 * Aborting the current (sub-)transaction as a whole has the right
2675 * semantics. We want all locks acquired in here to be released, not
2676 * reassigned to the parent and we do not want any database access
2677 * have persistent effects.
2678 */
2679 AbortCurrentTransaction();
2680
2681 /* make sure there's no cache pollution */
2683 {
2685 InvalidateSystemCaches();
2686 }
2687 else
2688 {
2691 txn->invalidations_distributed);
2692 }
2693
2695 {
2698 CurrentResourceOwner = cowner;
2699 }
2700
2701 /*
2702 * We are here due to one of the four reasons: 1. Decoding an
2703 * in-progress txn. 2. Decoding a prepared txn. 3. Decoding of a
2704 * prepared txn that was (partially) streamed. 4. Decoding a committed
2705 * txn.
2706 *
2707 * For 1, we allow truncation of txn data by removing the changes
2708 * already streamed but still keeping other things like invalidations,
2709 * snapshot, and tuplecids. For 2 and 3, we indicate
2710 * ReorderBufferTruncateTXN to do more elaborate truncation of txn
2711 * data as the entire transaction has been decoded except for commit.
2712 * For 4, as the entire txn has been decoded, we can fully clean up
2713 * the TXN reorder buffer.
2714 */
2716 {
2717 if (streaming)
2719
2721 /* Reset the CheckXidAlive */
2723 }
2724 else
2726 }
2727 PG_CATCH();
2728 {
2731
2732 /* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */
2735
2737
2738 /*
2739 * Force cache invalidation to happen outside of a valid transaction
2740 * to prevent catalog access as we just caught an error.
2741 */
2742 AbortCurrentTransaction();
2743
2744 /* make sure there's no cache pollution */
2746 {
2748 InvalidateSystemCaches();
2749 }
2750 else
2751 {
2754 txn->invalidations_distributed);
2755 }
2756
2758 {
2761 CurrentResourceOwner = cowner;
2762 }
2763
2764 /*
2765 * The error code ERRCODE_TRANSACTION_ROLLBACK indicates a concurrent
2766 * abort of the (sub)transaction we are streaming or preparing. We
2767 * need to do the cleanup and return gracefully on this error, see
2768 * SetupCheckXidLive.
2769 *
2770 * This error code can be thrown by one of the callbacks we call
2771 * during decoding so we need to ensure that we return gracefully only
2772 * when we are sending the data in streaming mode and the streaming is
2773 * not finished yet or when we are sending the data out on a PREPARE
2774 * during a two-phase commit.
2775 */
2778 {
2779 /* curtxn must be set for streaming or prepared transactions */
2781
2782 /* Cleanup the temporary error state. */
2783 FlushErrorState();
2786
2787 /* Remember the transaction is aborted. */
2790
2791 /* Mark the transaction is streamed if appropriate */
2794
2795 /* Reset the TXN so that it is allowed to stream remaining data. */
2797 command_id, prev_lsn,
2798 specinsert);
2799 }
2800 else
2801 {
2804 PG_RE_THROW();
2805 }
2806 }
2807 PG_END_TRY();
2808}
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 3293 of file reorderbuffer.c.
3294{
3295 /* many records won't have an xid assigned, centralize check here */
3298}
References fb(), InvalidTransactionId, and ReorderBufferTXNByXid().
Referenced by heap2_decode(), heap_decode(), LogicalDecodingProcessRecord(), logicalmsg_decode(), standby_decode(), xact_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 3481 of file reorderbuffer.c.
References ReorderBufferChange::action, ReorderBufferChange::data, fb(), ReorderBufferChange::inval, ReorderBufferChange::invalidations, 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, 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 2905 of file reorderbuffer.c.
2909{
2910 ReorderBufferTXN *txn;
2911
2913
2914 /* unknown transaction, nothing to do */
2916 return false;
2917
2918 /*
2919 * Remember the prepare information to be later used by commit prepared in
2920 * case we skip doing prepare.
2921 */
2922 txn->final_lsn = prepare_lsn;
2923 txn->end_lsn = end_lsn;
2924 txn->prepare_time = prepare_time;
2925 txn->origin_id = origin_id;
2926 txn->origin_lsn = origin_lsn;
2927
2928 /* Mark this transaction as a prepared transaction */
2931
2932 return true;
2933}
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 2821 of file reorderbuffer.c.
2826{
2827 Snapshot snapshot_now;
2829
2830 txn->final_lsn = commit_lsn;
2831 txn->end_lsn = end_lsn;
2832 txn->commit_time = commit_time;
2833 txn->origin_id = origin_id;
2834 txn->origin_lsn = origin_lsn;
2835
2836 /*
2837 * If the transaction was (partially) streamed, we need to commit it in a
2838 * 'streamed' way. That is, we first stream the remaining part of the
2839 * transaction, and then invoke stream_commit message.
2840 *
2841 * Called after everything (origin ID, LSN, ...) is stored in the
2842 * transaction to avoid passing that information directly.
2843 */
2845 {
2847 return;
2848 }
2849
2850 /*
2851 * If this transaction has no snapshot, it didn't make any changes to the
2852 * database, so there's nothing to decode. Note that
2853 * ReorderBufferCommitChild will have transferred any snapshots from
2854 * subtransactions if there were any.
2855 */
2857 {
2859
2860 /*
2861 * Removing this txn before a commit might result in the computation
2862 * of an incorrect restart_lsn. See SnapBuildProcessRunningXacts.
2863 */
2866 return;
2867 }
2868
2869 snapshot_now = txn->base_snapshot;
2870
2871 /* Process and send the changes to output plugin. */
2873 command_id, false);
2874}
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.
2171{
2172 /* Discard the changes that we just streamed */
2174
2175 /* Free all resources allocated for toast reconstruction */
2177
2178 /* Return the spec insert change if it is not NULL */
2180 {
2183 }
2184
2185 /*
2186 * For the streaming case, stop the stream and remember the command ID and
2187 * snapshot for the streaming run.
2188 */
2190 {
2193 }
2194
2195 /* All changes must be deallocated */
2197}
References Assert, fb(), rbtxn_is_prepared, rbtxn_is_streamed, ReorderBufferFreeChange(), ReorderBufferSaveTXNSnapshot(), ReorderBufferToastReset(), ReorderBufferTruncateTXN(), and ReorderBufferTXN::size.
Referenced by ReorderBufferProcessTXN().
◆ ReorderBufferRestoreChange()
|
static |
Definition at line 4685 of file reorderbuffer.c.
4687{
4688 ReorderBufferDiskChange *ondisk;
4689 ReorderBufferChange *change;
4690
4692
4694
4695 /* copy static part */
4697
4699
4700 /* restore individual stuff */
4702 {
4703 /* fall through these, they're all similar enough */
4709 {
4711
4714
4715 /* restore ->tuple */
4719
4720 /* reset t_data pointer into the new tuplebuf */
4723
4724 /* restore tuple data itself */
4727 }
4728
4730 {
4731 /* here, data might not be suitably aligned! */
4733
4736
4739
4740 /* restore ->tuple */
4744
4745 /* reset t_data pointer into the new tuplebuf */
4748
4749 /* restore tuple data itself */
4752 }
4753
4754 break;
4756 {
4757 Size prefix_size;
4758
4759 /* read prefix */
4763 prefix_size);
4766 data += prefix_size;
4767
4768 /* read the message */
4776
4777 break;
4778 }
4780 {
4783
4786
4787 /* read the message */
4789
4790 break;
4791 }
4793 {
4796 Size size;
4797
4799
4803
4805
4807
4813 break;
4814 }
4815 /* the base struct contains all the data, easy peasy */
4817 {
4818 Oid *relids;
4819
4823
4824 break;
4825 }
4830 break;
4831 }
4832
4834 txn->nentries_mem++;
4835
4836 /*
4837 * Update memory accounting for the restored change. We need to do this
4838 * although we don't check the memory limit when restoring the changes in
4839 * this branch (we only do that when initially queueing the changes after
4840 * decoding), because we will release the changes later, and that will
4841 * update the accounting too (subtracting the size from the counters). And
4842 * we don't want to underflow there.
4843 */
4845 ReorderBufferChangeSize(change));
4846}
References ReorderBufferChange::action, Assert, ReorderBufferDiskChange::change, ReorderBufferTXN::changes, ReorderBufferChange::data, data, dlist_push_tail(), fb(), HEAPTUPLESIZE, ReorderBufferChange::inval, ReorderBufferChange::invalidations, 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 4542 of file reorderbuffer.c.
4544{
4549
4552
4553 /* free current entries, so we have memory for more */
4555 {
4558
4561 }
4562 txn->nentries_mem = 0;
4564
4566
4568 {
4570 ReorderBufferDiskChange *ondisk;
4571
4572 CHECK_FOR_INTERRUPTS();
4573
4575 {
4577
4578 /* first time in */
4579 if (*segno == 0)
4581
4583
4584 /*
4585 * No need to care about TLIs here, only used during a single run,
4586 * so each LSN only maps to a specific WAL record.
4587 */
4589 *segno);
4590
4592
4593 /* No harm in resetting the offset even in case of failure */
4594 file->curOffset = 0;
4595
4597 {
4598 *fd = -1;
4599 (*segno)++;
4600 continue;
4601 }
4604 (errcode_for_file_access(),
4606 path)));
4607 }
4608
4609 /*
4610 * Read the statically sized part of a change which has information
4611 * about the total size. If we couldn't read a record, we're at the
4612 * end of this file.
4613 */
4618
4619 /* eof */
4621 {
4623 *fd = -1;
4624 (*segno)++;
4625 continue;
4626 }
4629 (errcode_for_file_access(),
4633 (errcode_for_file_access(),
4635 readBytes,
4637
4639
4641
4645
4649 file->curOffset,
4651
4654 (errcode_for_file_access(),
4658 (errcode_for_file_access(),
4660 readBytes,
4662
4664
4665 /*
4666 * ok, read a full change from disk, now restore it into proper
4667 * in-memory format
4668 */
4670 restored++;
4671 }
4672
4674}
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 4852 of file reorderbuffer.c.
4853{
4854 XLogSegNo first;
4856 XLogSegNo last;
4857
4860
4863
4864 /* iterate over all possible filenames, and delete them */
4866 {
4868
4872 (errcode_for_file_access(),
4874 }
4875}
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 4090 of file reorderbuffer.c.
4092{
4093 ReorderBufferDiskChange *ondisk;
4095
4097
4100
4102 {
4103 /* fall through these, they're all similar enough */
4108 {
4111 newtup;
4114
4117
4119 {
4123 }
4124
4126 {
4130 }
4131
4132 /* make sure we have enough space */
4134
4136 /* might have been reallocated above */
4138
4140 {
4143
4146 }
4147
4149 {
4152
4155 }
4156 break;
4157 }
4159 {
4162
4166
4168
4169 /* might have been reallocated above */
4171
4172 /* write the prefix including the size */
4176 prefix_size);
4177 data += prefix_size;
4178
4179 /* write the message including the size */
4185
4186 break;
4187 }
4189 {
4193
4195
4198
4199 /* might have been reallocated above */
4203
4204 break;
4205 }
4207 {
4210
4212
4216
4217 /* make sure we have enough space */
4220 /* might have been reallocated above */
4222
4225
4227 {
4231 }
4232
4234 {
4238 }
4239 break;
4240 }
4242 {
4243 Size size;
4245
4246 /* account for the OIDs of truncated relations */
4248 sz += size;
4249
4250 /* make sure we have enough space */
4252
4254 /* might have been reallocated above */
4256
4258 data += size;
4259
4260 break;
4261 }
4266 /* ReorderBufferChange contains everything important */
4267 break;
4268 }
4269
4270 ondisk->size = sz;
4271
4272 errno = 0;
4275 {
4277
4279
4280 /* if write didn't set errno, assume problem is no disk space */
4283 (errcode_for_file_access(),
4285 txn->xid)));
4286 }
4287 pgstat_report_wait_end();
4288
4289 /*
4290 * Keep the transaction's final_lsn up to date with each change we send to
4291 * disk, so that ReorderBufferRestoreCleanup works correctly. (We used to
4292 * only do this on commit and abort records, but that doesn't work if a
4293 * system crash leaves a transaction without its abort record).
4294 *
4295 * Make sure not to move it backwards.
4296 */
4299
4301}
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, 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 4921 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 3770 of file reorderbuffer.c.
3771{
3773 {
3776 }
3778 {
3781 }
3782}
References fb(), MemoryContextAlloc(), and repalloc().
Referenced by ReorderBufferRestoreChanges(), and ReorderBufferSerializeChange().
◆ ReorderBufferSerializeTXN()
|
static |
Definition at line 3995 of file reorderbuffer.c.
3996{
4001 Size spilled = 0;
4003
4006
4007 /* do the same to all child TXs */
4009 {
4011
4014 }
4015
4016 /* serialize changestream */
4018 {
4019 ReorderBufferChange *change;
4020
4022
4023 /*
4024 * store in segment in which it belongs by start lsn, don't split over
4025 * multiple segments tho
4026 */
4029 {
4031
4034
4036
4037 /*
4038 * No need to care about TLIs here, only used during a single run,
4039 * so each LSN only maps to a specific WAL record.
4040 */
4042 curOpenSegNo);
4043
4044 /* open segment, create it if necessary */
4047
4050 (errcode_for_file_access(),
4052 }
4053
4057
4058 spilled++;
4059 }
4060
4061 /* Update the memory counter */
4063
4064 /* update the statistics iff we have spilled anything */
4065 if (spilled)
4066 {
4067 rb->spillCount += 1;
4068 rb->spillBytes += size;
4069
4070 /* don't consider already serialized transactions */
4072
4073 /* update the decoding stats */
4075 }
4076
4079 txn->nentries_mem = 0;
4081
4084}
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 3324 of file reorderbuffer.c.
3326{
3327 ReorderBufferTXN *txn;
3329
3331
3332 /*
3333 * Fetch the transaction to operate on. If we know it's a subtransaction,
3334 * operate on its top-level transaction instead.
3335 */
3341
3343 txn->base_snapshot_lsn = lsn;
3345
3347}
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 2937 of file reorderbuffer.c.
2938{
2939 ReorderBufferTXN *txn;
2940
2942
2943 /* unknown transaction, nothing to do */
2945 return;
2946
2947 /* txn must have been marked as a prepared transaction */
2950}
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 4340 of file reorderbuffer.c.
4341{
4342 Snapshot snapshot_now;
4343 CommandId command_id;
4344 Size stream_bytes;
4346
4347 /* We can never reach here for a subtransaction. */
4349
4350 /*
4351 * We can't make any assumptions about base snapshot here, similar to what
4352 * ReorderBufferCommit() does. That relies on base_snapshot getting
4353 * transferred from subxact in ReorderBufferCommitChild(), but that was
4354 * not yet called as the transaction is in-progress.
4355 *
4356 * So just walk the subxacts and use the same logic here. But we only need
4357 * to do that once, when the transaction is streamed for the first time.
4358 * After that we need to reuse the snapshot from the previous run.
4359 *
4360 * Unlike DecodeCommit which adds xids of all the subtransactions in
4361 * snapshot's xip array via SnapBuildCommitTxn, we can't do that here but
4362 * we do add them to subxip array instead via ReorderBufferCopySnap. This
4363 * allows the catalog changes made in subtransactions decoded till now to
4364 * be visible.
4365 */
4367 {
4369
4370 /* make sure this transaction is streamed for the first time */
4372
4373 /* at the beginning we should have invalid command ID */
4375
4377 {
4379
4382 }
4383
4384 /*
4385 * If this transaction has no snapshot, it didn't make any changes to
4386 * the database till now, so there's nothing to decode.
4387 */
4389 {
4391 return;
4392 }
4393
4394 command_id = FirstCommandId;
4396 txn, command_id);
4397 }
4398 else
4399 {
4400 /* the transaction must have been already streamed */
4402
4403 /*
4404 * Nah, we already have snapshot from the previous streaming run. We
4405 * assume new subxacts can't move the LSN backwards, and so can't beat
4406 * the LSN condition in the previous branch (so no need to walk
4407 * through subxacts again). In fact, we must not do that as we may be
4408 * using snapshot half-way through the subxact.
4409 */
4410 command_id = txn->command_id;
4411
4412 /*
4413 * We can't use txn->snapshot_now directly because after the last
4414 * streaming run, we might have got some new sub-transactions. So we
4415 * need to add them to the snapshot.
4416 */
4418 txn, command_id);
4419
4420 /* Free the previously copied snapshot. */
4424 }
4425
4426 /*
4427 * Remember this information to be used later to update stats. We can't
4428 * update the stats here as an error while processing the changes would
4429 * lead to the accumulation of stats even though we haven't streamed all
4430 * the changes.
4431 */
4433 stream_bytes = txn->total_size;
4434
4435 /* Process and send the changes to output plugin. */
4437 command_id, true);
4438
4439 rb->streamCount += 1;
4440 rb->streamBytes += stream_bytes;
4441
4442 /* Don't consider already streamed transaction. */
4444
4445 /* update the decoding stats */
4447
4451}
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 4993 of file reorderbuffer.c.
4995{
4998 bool found;
5000 bool isnull;
5001 Pointer chunk;
5003 Oid chunk_id;
5005
5008
5010
5013 Assert(!isnull);
5015 Assert(!isnull);
5016
5019
5020 if (!found)
5021 {
5023 ent->num_chunks = 0;
5024 ent->last_chunk_seq = 0;
5025 ent->size = 0;
5028
5031 chunk_seq, chunk_id);
5032 }
5036
5038 Assert(!isnull);
5039
5040 /* calculate size so we can allocate the right size at once later */
5044 /* could happen due to heap_form_tuple doing its thing */
5046 else
5048
5051 ent->num_chunks++;
5053}
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 4973 of file reorderbuffer.c.
4974{
4976
4978
4984}
References Assert, fb(), HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, and ReorderBufferTXN::toast_hash.
Referenced by ReorderBufferToastAppendChunk().
◆ ReorderBufferToastReplace()
|
static |
Definition at line 5076 of file reorderbuffer.c.
5078{
5079 TupleDesc desc;
5082 bool *isnull;
5085 Relation toast_rel;
5087 MemoryContext oldcontext;
5090
5091 /* no toast tuples changed */
5093 return;
5094
5095 /*
5096 * We're going to modify the size of the change. So, to make sure the
5097 * accounting is correct we record the current change size and then after
5098 * re-computing the change we'll subtract the recorded size and then
5099 * re-add the new change size at the end. We don't immediately subtract
5100 * the old size because if there is any error before we add the new size,
5101 * we will release the changes and that will update the accounting info
5102 * (subtracting the size from the counters). And we don't want to
5103 * underflow there.
5104 */
5106
5108
5109 /* we should only have toast tuples in an INSERT or UPDATE */
5111
5112 desc = RelationGetDescr(relation);
5113
5118
5120
5121 /* should we allocate from stack instead? */
5125
5127
5129
5131 {
5135
5136 /* va_rawsize is the size of the original datum -- including header */
5137 varatt_external toast_pointer;
5140 varlena *reconstructed;
5143
5145 continue;
5146
5147 /* not a varlena datatype */
5149 continue;
5150
5151 /* no data */
5153 continue;
5154
5155 /* ok, we know we have a toast datum */
5157
5158 /* no need to do anything if the tuple isn't external */
5160 continue;
5161
5163
5164 /*
5165 * Check whether the toast tuple changed, replace if so.
5166 */
5169 &toast_pointer.va_valueid,
5170 HASH_FIND,
5171 NULL);
5173 continue;
5174
5175 new_datum =
5177
5179
5181
5182 ent->reconstructed = reconstructed;
5183
5184 /* stitch toast tuple back together from its parts */
5186 {
5190 Pointer chunk;
5191
5195
5199
5201 VARDATA(chunk),
5204 }
5206
5207 /* make sure its marked as compressed or not */
5210 else
5212
5214 redirect_pointer.pointer = reconstructed;
5215
5219
5221 }
5222
5223 /*
5224 * Build tuple in separate memory & copy tuple back into the tuplebuf
5225 * passed to the output plugin. We can't directly heap_fill_tuple() into
5226 * the tuplebuf because attrs[] will point back into the current content.
5227 */
5231
5234
5235 /*
5236 * free resources we won't further need, more persistent stuff will be
5237 * free'd in ReorderBufferToastReset().
5238 */
5239 RelationClose(toast_rel);
5242 {
5245 }
5248 pfree(isnull);
5249
5250 MemoryContextSwitchTo(oldcontext);
5251
5252 /* subtract the old change size */
5254 /* now add the change back, with the correct size */
5256 ReorderBufferChangeSize(change));
5257}
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, 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 5263 of file reorderbuffer.c.
5264{
5267
5269 return;
5270
5271 /* sequentially walk over the hash and free everything */
5274 {
5276
5279
5281 {
5282 ReorderBufferChange *change =
5284
5287 }
5288 }
5289
5292}
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 3787 of file reorderbuffer.c.
3788{
3791
3793 return -1;
3795 return 1;
3796 return 0;
3797}
References a, b, fb(), and pairingheap_const_container.
Referenced by ReorderBufferAllocate().
◆ ReorderBufferXidHasBaseSnapshot()
| bool ReorderBufferXidHasBaseSnapshot | ( | ReorderBuffer * | rb, |
| TransactionId | xid | ||
| ) |
Definition at line 3740 of file reorderbuffer.c.
3741{
3742 ReorderBufferTXN *txn;
3743
3746
3747 /* transaction isn't known yet, ergo no snapshot */
3749 return false;
3750
3751 /* a known subtxn? operate on top-level txn instead */
3755
3757}
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 3723 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 3651 of file reorderbuffer.c.
3653{
3654 ReorderBufferTXN *txn;
3655
3657
3659 {
3662 }
3663
3664 /*
3665 * Mark top-level transaction as having catalog changes too if one of its
3666 * children has so that the ReorderBufferBuildTupleCidHash can
3667 * conveniently check just top-level transaction and decide whether to
3668 * build the hash table or not.
3669 */
3671 {
3673
3675 {
3678 }
3679 }
3680}
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 5555 of file reorderbuffer.c.
5559{
5562 ForkNumber forkno;
5563 BlockNumber blockno;
5565
5566 /*
5567 * Return unresolved if tuplecid_data is not valid. That's because when
5568 * streaming in-progress transactions we may run into tuples with the CID
5569 * before actually decoding them. Think e.g. about INSERT followed by
5570 * TRUNCATE, where the TRUNCATE may not be decoded yet when applying the
5571 * INSERT. So in such cases, we assume the CID is from the future
5572 * command.
5573 */
5575 return false;
5576
5577 /* be careful about padding */
5579
5581
5582 /*
5583 * get relfilelocator from the buffer, no convenient way to access it
5584 * other than that.
5585 */
5587
5588 /* tuples can only be in the main fork */
5591
5593 &key.tid);
5594
5595restart:
5598
5599 /*
5600 * failed to find a mapping, check whether the table was rewritten and
5601 * apply mapping if so, but only do that once - there can be no new
5602 * mappings while we are in here since we have to hold a lock on the
5603 * relation.
5604 */
5606 {
5608 /* now check but don't update for a mapping again */
5610 goto restart;
5611 }
5613 return false;
5614
5615 if (cmin)
5616 *cmin = ent->cmin;
5617 if (cmax)
5618 *cmax = ent->cmax;
5619 return true;
5620}
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 4939 of file reorderbuffer.c.
4940{
4943
4946 {
4949 continue;
4950
4951 /* if it cannot be a slot, skip the directory */
4953 continue;
4954
4955 /*
4956 * ok, has to be a surviving logical slot, iterate and delete
4957 * everything starting with xid-*
4958 */
4960 }
4962}
References AllocateDir(), DEBUG2, fb(), FreeDir(), PG_REPLSLOT_DIR, ReadDir(), ReorderBufferCleanupSerializedTXNs(), and ReplicationSlotValidateName().
Referenced by StartupXLOG().
◆ TransactionIdInArray()
|
static |
Definition at line 5454 of file reorderbuffer.c.
References fb(), and xidComparator().
Referenced by UpdateLogicalMappings().
◆ UpdateLogicalMappings()
Definition at line 5477 of file reorderbuffer.c.
5478{
5482 ListCell *file;
5484
5487 {
5494 f_lo;
5495 RewriteMappingFile *f;
5496
5499 continue;
5500
5501 /* Ignore files that aren't ours */
5503 continue;
5504
5509
5511
5512 /* mapping for another database */
5514 continue;
5515
5516 /* mapping for another relation */
5518 continue;
5519
5520 /* did the creating transaction abort? */
5522 continue;
5523
5524 /* not for our transaction */
5526 continue;
5527
5528 /* ok, relevant, queue for apply */
5532 files = lappend(files, f);
5533 }
5535
5536 /* sort files so we apply them in LSN order */
5538
5539 foreach(file, files)
5540 {
5542
5544 snapshot->subxip[0]);
5546 pfree(f);
5547 }
5548}
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().
