granicus.if.org Git - postgresql/blob - src/backend/replication/logical/reorderbuffer.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * reorderbuffer.c
   4  *        PostgreSQL logical replay/reorder buffer management
   5  *
   6  *
   7  * Copyright (c) 2012-2016, PostgreSQL Global Development Group
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        src/backend/replication/reorderbuffer.c
  12  *
  13  * NOTES
  14  *        This module gets handed individual pieces of transactions in the order
  15  *        they are written to the WAL and is responsible to reassemble them into
  16  *        toplevel transaction sized pieces. When a transaction is completely
  17  *        reassembled - signalled by reading the transaction commit record - it
  18  *        will then call the output plugin (c.f. ReorderBufferCommit()) with the
  19  *        individual changes. The output plugins rely on snapshots built by
  20  *        snapbuild.c which hands them to us.
  21  *
  22  *        Transactions and subtransactions/savepoints in postgres are not
  23  *        immediately linked to each other from outside the performing
  24  *        backend. Only at commit/abort (or special xact_assignment records) they
  25  *        are linked together. Which means that we will have to splice together a
  26  *        toplevel transaction from its subtransactions. To do that efficiently we
  27  *        build a binary heap indexed by the smallest current lsn of the individual
  28  *        subtransactions' changestreams. As the individual streams are inherently
  29  *        ordered by LSN - since that is where we build them from - the transaction
  30  *        can easily be reassembled by always using the subtransaction with the
  31  *        smallest current LSN from the heap.
  32  *
  33  *        In order to cope with large transactions - which can be several times as
  34  *        big as the available memory - this module supports spooling the contents
  35  *        of a large transactions to disk. When the transaction is replayed the
  36  *        contents of individual (sub-)transactions will be read from disk in
  37  *        chunks.
  38  *
  39  *        This module also has to deal with reassembling toast records from the
  40  *        individual chunks stored in WAL. When a new (or initial) version of a
  41  *        tuple is stored in WAL it will always be preceded by the toast chunks
  42  *        emitted for the columns stored out of line. Within a single toplevel
  43  *        transaction there will be no other data carrying records between a row's
  44  *        toast chunks and the row data itself. See ReorderBufferToast* for
  45  *        details.
  46  * -------------------------------------------------------------------------
  47  */
  48 #include "postgres.h"
  49
  50 #include <unistd.h>
  51 #include <sys/stat.h>
  52
  53 #include "access/rewriteheap.h"
  54 #include "access/transam.h"
  55 #include "access/tuptoaster.h"
  56 #include "access/xact.h"
  57 #include "access/xlog_internal.h"
  58 #include "catalog/catalog.h"
  59 #include "lib/binaryheap.h"
  60 #include "miscadmin.h"
  61 #include "replication/logical.h"
  62 #include "replication/reorderbuffer.h"
  63 #include "replication/slot.h"
  64 #include "replication/snapbuild.h"              /* just for SnapBuildSnapDecRefcount */
  65 #include "storage/bufmgr.h"
  66 #include "storage/fd.h"
  67 #include "storage/sinval.h"
  68 #include "utils/builtins.h"
  69 #include "utils/combocid.h"
  70 #include "utils/memdebug.h"
  71 #include "utils/memutils.h"
  72 #include "utils/rel.h"
  73 #include "utils/relfilenodemap.h"
  74 #include "utils/tqual.h"
  75
  76
  77 /* entry for a hash table we use to map from xid to our transaction state */
  78 typedef struct ReorderBufferTXNByIdEnt
  79 {
  80         TransactionId xid;
  81         ReorderBufferTXN *txn;
  82 } ReorderBufferTXNByIdEnt;
  83
  84 /* data structures for (relfilenode, ctid) => (cmin, cmax) mapping */
  85 typedef struct ReorderBufferTupleCidKey
  86 {
  87         RelFileNode relnode;
  88         ItemPointerData tid;
  89 } ReorderBufferTupleCidKey;
  90
  91 typedef struct ReorderBufferTupleCidEnt
  92 {
  93         ReorderBufferTupleCidKey key;
  94         CommandId       cmin;
  95         CommandId       cmax;
  96         CommandId       combocid;               /* just for debugging */
  97 } ReorderBufferTupleCidEnt;
  98
  99 /* k-way in-order change iteration support structures */
 100 typedef struct ReorderBufferIterTXNEntry
 101 {
 102         XLogRecPtr      lsn;
 103         ReorderBufferChange *change;
 104         ReorderBufferTXN *txn;
 105         int                     fd;
 106         XLogSegNo       segno;
 107 } ReorderBufferIterTXNEntry;
 108
 109 typedef struct ReorderBufferIterTXNState
 110 {
 111         binaryheap *heap;
 112         Size            nr_txns;
 113         dlist_head      old_change;
 114         ReorderBufferIterTXNEntry entries[FLEXIBLE_ARRAY_MEMBER];
 115 } ReorderBufferIterTXNState;
 116
 117 /* toast datastructures */
 118 typedef struct ReorderBufferToastEnt
 119 {
 120         Oid                     chunk_id;               /* toast_table.chunk_id */
 121         int32           last_chunk_seq; /* toast_table.chunk_seq of the last chunk we
 122                                                                  * have seen */
 123         Size            num_chunks;             /* number of chunks we've already seen */
 124         Size            size;                   /* combined size of chunks seen */
 125         dlist_head      chunks;                 /* linked list of chunks */
 126         struct varlena *reconstructed;          /* reconstructed varlena now pointed
 127                                                                                  * to in main tup */
 128 } ReorderBufferToastEnt;
 129
 130 /* Disk serialization support datastructures */
 131 typedef struct ReorderBufferDiskChange
 132 {
 133         Size            size;
 134         ReorderBufferChange change;
 135         /* data follows */
 136 } ReorderBufferDiskChange;
 137
 138 /*
 139  * Maximum number of changes kept in memory, per transaction. After that,
 140  * changes are spooled to disk.
 141  *
 142  * The current value should be sufficient to decode the entire transaction
 143  * without hitting disk in OLTP workloads, while starting to spool to disk in
 144  * other workloads reasonably fast.
 145  *
 146  * At some point in the future it probably makes sense to have a more elaborate
 147  * resource management here, but it's not entirely clear what that would look
 148  * like.
 149  */
 150 static const Size max_changes_in_memory = 4096;
 151
 152 /*
 153  * We use a very simple form of a slab allocator for frequently allocated
 154  * objects, simply keeping a fixed number in a linked list when unused,
 155  * instead pfree()ing them. Without that in many workloads aset.c becomes a
 156  * major bottleneck, especially when spilling to disk while decoding batch
 157  * workloads.
 158  */
 159 static const Size max_cached_changes = 4096 * 2;
 160 static const Size max_cached_tuplebufs = 4096 * 2;              /* ~8MB */
 161 static const Size max_cached_transactions = 512;
 162
 163
 164 /* ---------------------------------------
 165  * primary reorderbuffer support routines
 166  * ---------------------------------------
 167  */
 168 static ReorderBufferTXN *ReorderBufferGetTXN(ReorderBuffer *rb);
 169 static void ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
 170 static ReorderBufferTXN *ReorderBufferTXNByXid(ReorderBuffer *rb,
 171                                           TransactionId xid, bool create, bool *is_new,
 172                                           XLogRecPtr lsn, bool create_as_top);
 173
 174 static void AssertTXNLsnOrder(ReorderBuffer *rb);
 175
 176 /* ---------------------------------------
 177  * support functions for lsn-order iterating over the ->changes of a
 178  * transaction and its subtransactions
 179  *
 180  * used for iteration over the k-way heap merge of a transaction and its
 181  * subtransactions
 182  * ---------------------------------------
 183  */
 184 static ReorderBufferIterTXNState *ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn);
 185 static ReorderBufferChange *
 186                         ReorderBufferIterTXNNext(ReorderBuffer *rb, ReorderBufferIterTXNState *state);
 187 static void ReorderBufferIterTXNFinish(ReorderBuffer *rb,
 188                                                    ReorderBufferIterTXNState *state);
 189 static void ReorderBufferExecuteInvalidations(ReorderBuffer *rb, ReorderBufferTXN *txn);
 190
 191 /*
 192  * ---------------------------------------
 193  * Disk serialization support functions
 194  * ---------------------------------------
 195  */
 196 static void ReorderBufferCheckSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
 197 static void ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
 198 static void ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 199                                                          int fd, ReorderBufferChange *change);
 200 static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 201                                                         int *fd, XLogSegNo *segno);
 202 static void ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 203                                                    char *change);
 204 static void ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn);
 205
 206 static void ReorderBufferFreeSnap(ReorderBuffer *rb, Snapshot snap);
 207 static Snapshot ReorderBufferCopySnap(ReorderBuffer *rb, Snapshot orig_snap,
 208                                           ReorderBufferTXN *txn, CommandId cid);
 209
 210 /* ---------------------------------------
 211  * toast reassembly support
 212  * ---------------------------------------
 213  */
 214 static void ReorderBufferToastInitHash(ReorderBuffer *rb, ReorderBufferTXN *txn);
 215 static void ReorderBufferToastReset(ReorderBuffer *rb, ReorderBufferTXN *txn);
 216 static void ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 217                                                   Relation relation, ReorderBufferChange *change);
 218 static void ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
 219                                                           Relation relation, ReorderBufferChange *change);
 220
 221
 222 /*
 223  * Allocate a new ReorderBuffer
 224  */
 225 ReorderBuffer *
 226 ReorderBufferAllocate(void)
 227 {
 228         ReorderBuffer *buffer;
 229         HASHCTL         hash_ctl;
 230         MemoryContext new_ctx;
 231
 232         /* allocate memory in own context, to have better accountability */
 233         new_ctx = AllocSetContextCreate(CurrentMemoryContext,
 234                                                                         "ReorderBuffer",
 235                                                                         ALLOCSET_DEFAULT_MINSIZE,
 236                                                                         ALLOCSET_DEFAULT_INITSIZE,
 237                                                                         ALLOCSET_DEFAULT_MAXSIZE);
 238
 239         buffer =
 240                 (ReorderBuffer *) MemoryContextAlloc(new_ctx, sizeof(ReorderBuffer));
 241
 242         memset(&hash_ctl, 0, sizeof(hash_ctl));
 243
 244         buffer->context = new_ctx;
 245
 246         hash_ctl.keysize = sizeof(TransactionId);
 247         hash_ctl.entrysize = sizeof(ReorderBufferTXNByIdEnt);
 248         hash_ctl.hcxt = buffer->context;
 249
 250         buffer->by_txn = hash_create("ReorderBufferByXid", 1000, &hash_ctl,
 251                                                                  HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
 252
 253         buffer->by_txn_last_xid = InvalidTransactionId;
 254         buffer->by_txn_last_txn = NULL;
 255
 256         buffer->nr_cached_transactions = 0;
 257         buffer->nr_cached_changes = 0;
 258         buffer->nr_cached_tuplebufs = 0;
 259
 260         buffer->outbuf = NULL;
 261         buffer->outbufsize = 0;
 262
 263         buffer->current_restart_decoding_lsn = InvalidXLogRecPtr;
 264
 265         dlist_init(&buffer->toplevel_by_lsn);
 266         dlist_init(&buffer->cached_transactions);
 267         dlist_init(&buffer->cached_changes);
 268         slist_init(&buffer->cached_tuplebufs);
 269
 270         return buffer;
 271 }
 272
 273 /*
 274  * Free a ReorderBuffer
 275  */
 276 void
 277 ReorderBufferFree(ReorderBuffer *rb)
 278 {
 279         MemoryContext context = rb->context;
 280
 281         /*
 282          * We free separately allocated data by entirely scrapping reorderbuffer's
 283          * memory context.
 284          */
 285         MemoryContextDelete(context);
 286 }
 287
 288 /*
 289  * Get an unused, possibly preallocated, ReorderBufferTXN.
 290  */
 291 static ReorderBufferTXN *
 292 ReorderBufferGetTXN(ReorderBuffer *rb)
 293 {
 294         ReorderBufferTXN *txn;
 295
 296         /* check the slab cache */
 297         if (rb->nr_cached_transactions > 0)
 298         {
 299                 rb->nr_cached_transactions--;
 300                 txn = (ReorderBufferTXN *)
 301                         dlist_container(ReorderBufferTXN, node,
 302                                                         dlist_pop_head_node(&rb->cached_transactions));
 303         }
 304         else
 305         {
 306                 txn = (ReorderBufferTXN *)
 307                         MemoryContextAlloc(rb->context, sizeof(ReorderBufferTXN));
 308         }
 309
 310         memset(txn, 0, sizeof(ReorderBufferTXN));
 311
 312         dlist_init(&txn->changes);
 313         dlist_init(&txn->tuplecids);
 314         dlist_init(&txn->subtxns);
 315
 316         return txn;
 317 }
 318
 319 /*
 320  * Free a ReorderBufferTXN.
 321  *
 322  * Deallocation might be delayed for efficiency purposes, for details check
 323  * the comments above max_cached_changes's definition.
 324  */
 325 static void
 326 ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 327 {
 328         /* clean the lookup cache if we were cached (quite likely) */
 329         if (rb->by_txn_last_xid == txn->xid)
 330         {
 331                 rb->by_txn_last_xid = InvalidTransactionId;
 332                 rb->by_txn_last_txn = NULL;
 333         }
 334
 335         /* free data that's contained */
 336
 337         if (txn->tuplecid_hash != NULL)
 338         {
 339                 hash_destroy(txn->tuplecid_hash);
 340                 txn->tuplecid_hash = NULL;
 341         }
 342
 343         if (txn->invalidations)
 344         {
 345                 pfree(txn->invalidations);
 346                 txn->invalidations = NULL;
 347         }
 348
 349         /* check whether to put into the slab cache */
 350         if (rb->nr_cached_transactions < max_cached_transactions)
 351         {
 352                 rb->nr_cached_transactions++;
 353                 dlist_push_head(&rb->cached_transactions, &txn->node);
 354                 VALGRIND_MAKE_MEM_UNDEFINED(txn, sizeof(ReorderBufferTXN));
 355                 VALGRIND_MAKE_MEM_DEFINED(&txn->node, sizeof(txn->node));
 356         }
 357         else
 358         {
 359                 pfree(txn);
 360         }
 361 }
 362
 363 /*
 364  * Get an unused, possibly preallocated, ReorderBufferChange.
 365  */
 366 ReorderBufferChange *
 367 ReorderBufferGetChange(ReorderBuffer *rb)
 368 {
 369         ReorderBufferChange *change;
 370
 371         /* check the slab cache */
 372         if (rb->nr_cached_changes)
 373         {
 374                 rb->nr_cached_changes--;
 375                 change = (ReorderBufferChange *)
 376                         dlist_container(ReorderBufferChange, node,
 377                                                         dlist_pop_head_node(&rb->cached_changes));
 378         }
 379         else
 380         {
 381                 change = (ReorderBufferChange *)
 382                         MemoryContextAlloc(rb->context, sizeof(ReorderBufferChange));
 383         }
 384
 385         memset(change, 0, sizeof(ReorderBufferChange));
 386         return change;
 387 }
 388
 389 /*
 390  * Free an ReorderBufferChange.
 391  *
 392  * Deallocation might be delayed for efficiency purposes, for details check
 393  * the comments above max_cached_changes's definition.
 394  */
 395 void
 396 ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change)
 397 {
 398         /* free contained data */
 399         switch (change->action)
 400         {
 401                 case REORDER_BUFFER_CHANGE_INSERT:
 402                 case REORDER_BUFFER_CHANGE_UPDATE:
 403                 case REORDER_BUFFER_CHANGE_DELETE:
 404                 case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
 405                         if (change->data.tp.newtuple)
 406                         {
 407                                 ReorderBufferReturnTupleBuf(rb, change->data.tp.newtuple);
 408                                 change->data.tp.newtuple = NULL;
 409                         }
 410
 411                         if (change->data.tp.oldtuple)
 412                         {
 413                                 ReorderBufferReturnTupleBuf(rb, change->data.tp.oldtuple);
 414                                 change->data.tp.oldtuple = NULL;
 415                         }
 416                         break;
 417                 case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
 418                         if (change->data.snapshot)
 419                         {
 420                                 ReorderBufferFreeSnap(rb, change->data.snapshot);
 421                                 change->data.snapshot = NULL;
 422                         }
 423                         break;
 424                         /* no data in addition to the struct itself */
 425                 case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
 426                 case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
 427                 case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
 428                         break;
 429         }
 430
 431         /* check whether to put into the slab cache */
 432         if (rb->nr_cached_changes < max_cached_changes)
 433         {
 434                 rb->nr_cached_changes++;
 435                 dlist_push_head(&rb->cached_changes, &change->node);
 436                 VALGRIND_MAKE_MEM_UNDEFINED(change, sizeof(ReorderBufferChange));
 437                 VALGRIND_MAKE_MEM_DEFINED(&change->node, sizeof(change->node));
 438         }
 439         else
 440         {
 441                 pfree(change);
 442         }
 443 }
 444
 445
 446 /*
 447  * Get an unused, possibly preallocated, ReorderBufferTupleBuf fitting at
 448  * least a tuple of size tuple_len (excluding header overhead).
 449  */
 450 ReorderBufferTupleBuf *
 451 ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
 452 {
 453         ReorderBufferTupleBuf *tuple;
 454         Size            alloc_len;
 455
 456         alloc_len = tuple_len + SizeofHeapTupleHeader;
 457
 458         /*
 459          * Most tuples are below MaxHeapTupleSize, so we use a slab allocator for
 460          * those. Thus always allocate at least MaxHeapTupleSize. Note that tuples
 461          * tuples generated for oldtuples can be bigger, as they don't have
 462          * out-of-line toast columns.
 463          */
 464         if (alloc_len < MaxHeapTupleSize)
 465                 alloc_len = MaxHeapTupleSize;
 466
 467
 468         /* if small enough, check the slab cache */
 469         if (alloc_len <= MaxHeapTupleSize && rb->nr_cached_tuplebufs)
 470         {
 471                 rb->nr_cached_tuplebufs--;
 472                 tuple = slist_container(ReorderBufferTupleBuf, node,
 473                                                                 slist_pop_head_node(&rb->cached_tuplebufs));
 474 #ifdef USE_ASSERT_CHECKING
 475                 memset(&tuple->tuple, 0xa9, sizeof(HeapTupleData));
 476 #endif
 477                 tuple->tuple.t_data = ReorderBufferTupleBufData(tuple);
 478 #ifdef USE_ASSERT_CHECKING
 479                 memset(tuple->tuple.t_data, 0xa8, tuple->alloc_tuple_size);
 480 #endif
 481         }
 482         else
 483         {
 484                 tuple = (ReorderBufferTupleBuf *)
 485                         MemoryContextAlloc(rb->context,
 486                                                            sizeof(ReorderBufferTupleBuf) + alloc_len);
 487                 tuple->alloc_tuple_size = alloc_len;
 488                 tuple->tuple.t_data = ReorderBufferTupleBufData(tuple);
 489         }
 490
 491         return tuple;
 492 }
 493
 494 /*
 495  * Free an ReorderBufferTupleBuf.
 496  *
 497  * Deallocation might be delayed for efficiency purposes, for details check
 498  * the comments above max_cached_changes's definition.
 499  */
 500 void
 501 ReorderBufferReturnTupleBuf(ReorderBuffer *rb, ReorderBufferTupleBuf *tuple)
 502 {
 503         /* check whether to put into the slab cache, oversized tuples never are */
 504         if (tuple->alloc_tuple_size == MaxHeapTupleSize &&
 505                 rb->nr_cached_tuplebufs < max_cached_tuplebufs)
 506         {
 507                 rb->nr_cached_tuplebufs++;
 508                 slist_push_head(&rb->cached_tuplebufs, &tuple->node);
 509                 VALGRIND_MAKE_MEM_UNDEFINED(tuple->tuple.t_data, tuple->alloc_tuple_size);
 510                 VALGRIND_MAKE_MEM_UNDEFINED(tuple, sizeof(ReorderBufferTupleBuf));
 511                 VALGRIND_MAKE_MEM_DEFINED(&tuple->node, sizeof(tuple->node));
 512                 VALGRIND_MAKE_MEM_DEFINED(&tuple->alloc_tuple_size, sizeof(tuple->alloc_tuple_size));
 513         }
 514         else
 515         {
 516                 pfree(tuple);
 517         }
 518 }
 519
 520 /*
 521  * Return the ReorderBufferTXN from the given buffer, specified by Xid.
 522  * If create is true, and a transaction doesn't already exist, create it
 523  * (with the given LSN, and as top transaction if that's specified);
 524  * when this happens, is_new is set to true.
 525  */
 526 static ReorderBufferTXN *
 527 ReorderBufferTXNByXid(ReorderBuffer *rb, TransactionId xid, bool create,
 528                                           bool *is_new, XLogRecPtr lsn, bool create_as_top)
 529 {
 530         ReorderBufferTXN *txn;
 531         ReorderBufferTXNByIdEnt *ent;
 532         bool            found;
 533
 534         Assert(TransactionIdIsValid(xid));
 535         Assert(!create || lsn != InvalidXLogRecPtr);
 536
 537         /*
 538          * Check the one-entry lookup cache first
 539          */
 540         if (TransactionIdIsValid(rb->by_txn_last_xid) &&
 541                 rb->by_txn_last_xid == xid)
 542         {
 543                 txn = rb->by_txn_last_txn;
 544
 545                 if (txn != NULL)
 546                 {
 547                         /* found it, and it's valid */
 548                         if (is_new)
 549                                 *is_new = false;
 550                         return txn;
 551                 }
 552
 553                 /*
 554                  * cached as non-existant, and asked not to create? Then nothing else
 555                  * to do.
 556                  */
 557                 if (!create)
 558                         return NULL;
 559                 /* otherwise fall through to create it */
 560         }
 561
 562         /*
 563          * If the cache wasn't hit or it yielded an "does-not-exist" and we want
 564          * to create an entry.
 565          */
 566
 567         /* search the lookup table */
 568         ent = (ReorderBufferTXNByIdEnt *)
 569                 hash_search(rb->by_txn,
 570                                         (void *) &xid,
 571                                         create ? HASH_ENTER : HASH_FIND,
 572                                         &found);
 573         if (found)
 574                 txn = ent->txn;
 575         else if (create)
 576         {
 577                 /* initialize the new entry, if creation was requested */
 578                 Assert(ent != NULL);
 579
 580                 ent->txn = ReorderBufferGetTXN(rb);
 581                 ent->txn->xid = xid;
 582                 txn = ent->txn;
 583                 txn->first_lsn = lsn;
 584                 txn->restart_decoding_lsn = rb->current_restart_decoding_lsn;
 585
 586                 if (create_as_top)
 587                 {
 588                         dlist_push_tail(&rb->toplevel_by_lsn, &txn->node);
 589                         AssertTXNLsnOrder(rb);
 590                 }
 591         }
 592         else
 593                 txn = NULL;                             /* not found and not asked to create */
 594
 595         /* update cache */
 596         rb->by_txn_last_xid = xid;
 597         rb->by_txn_last_txn = txn;
 598
 599         if (is_new)
 600                 *is_new = !found;
 601
 602         Assert(!create || !!txn);
 603         return txn;
 604 }
 605
 606 /*
 607  * Queue a change into a transaction so it can be replayed upon commit.
 608  */
 609 void
 610 ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
 611                                                  ReorderBufferChange *change)
 612 {
 613         ReorderBufferTXN *txn;
 614
 615         txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 616
 617         change->lsn = lsn;
 618         Assert(InvalidXLogRecPtr != lsn);
 619         dlist_push_tail(&txn->changes, &change->node);
 620         txn->nentries++;
 621         txn->nentries_mem++;
 622
 623         ReorderBufferCheckSerializeTXN(rb, txn);
 624 }
 625
 626 static void
 627 AssertTXNLsnOrder(ReorderBuffer *rb)
 628 {
 629 #ifdef USE_ASSERT_CHECKING
 630         dlist_iter      iter;
 631         XLogRecPtr      prev_first_lsn = InvalidXLogRecPtr;
 632
 633         dlist_foreach(iter, &rb->toplevel_by_lsn)
 634         {
 635                 ReorderBufferTXN *cur_txn;
 636
 637                 cur_txn = dlist_container(ReorderBufferTXN, node, iter.cur);
 638                 Assert(cur_txn->first_lsn != InvalidXLogRecPtr);
 639
 640                 if (cur_txn->end_lsn != InvalidXLogRecPtr)
 641                         Assert(cur_txn->first_lsn <= cur_txn->end_lsn);
 642
 643                 if (prev_first_lsn != InvalidXLogRecPtr)
 644                         Assert(prev_first_lsn < cur_txn->first_lsn);
 645
 646                 Assert(!cur_txn->is_known_as_subxact);
 647                 prev_first_lsn = cur_txn->first_lsn;
 648         }
 649 #endif
 650 }
 651
 652 ReorderBufferTXN *
 653 ReorderBufferGetOldestTXN(ReorderBuffer *rb)
 654 {
 655         ReorderBufferTXN *txn;
 656
 657         if (dlist_is_empty(&rb->toplevel_by_lsn))
 658                 return NULL;
 659
 660         AssertTXNLsnOrder(rb);
 661
 662         txn = dlist_head_element(ReorderBufferTXN, node, &rb->toplevel_by_lsn);
 663
 664         Assert(!txn->is_known_as_subxact);
 665         Assert(txn->first_lsn != InvalidXLogRecPtr);
 666         return txn;
 667 }
 668
 669 void
 670 ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
 671 {
 672         rb->current_restart_decoding_lsn = ptr;
 673 }
 674
 675 void
 676 ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid,
 677                                                  TransactionId subxid, XLogRecPtr lsn)
 678 {
 679         ReorderBufferTXN *txn;
 680         ReorderBufferTXN *subtxn;
 681         bool            new_top;
 682         bool            new_sub;
 683
 684         txn = ReorderBufferTXNByXid(rb, xid, true, &new_top, lsn, true);
 685         subtxn = ReorderBufferTXNByXid(rb, subxid, true, &new_sub, lsn, false);
 686
 687         if (new_sub)
 688         {
 689                 /*
 690                  * we assign subtransactions to top level transaction even if we don't
 691                  * have data for it yet, assignment records frequently reference xids
 692                  * that have not yet produced any records. Knowing those aren't top
 693                  * level xids allows us to make processing cheaper in some places.
 694                  */
 695                 dlist_push_tail(&txn->subtxns, &subtxn->node);
 696                 txn->nsubtxns++;
 697         }
 698         else if (!subtxn->is_known_as_subxact)
 699         {
 700                 subtxn->is_known_as_subxact = true;
 701                 Assert(subtxn->nsubtxns == 0);
 702
 703                 /* remove from lsn order list of top-level transactions */
 704                 dlist_delete(&subtxn->node);
 705
 706                 /* add to toplevel transaction */
 707                 dlist_push_tail(&txn->subtxns, &subtxn->node);
 708                 txn->nsubtxns++;
 709         }
 710         else if (new_top)
 711         {
 712                 elog(ERROR, "existing subxact assigned to unknown toplevel xact");
 713         }
 714 }
 715
 716 /*
 717  * Associate a subtransaction with its toplevel transaction at commit
 718  * time. There may be no further changes added after this.
 719  */
 720 void
 721 ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid,
 722                                                  TransactionId subxid, XLogRecPtr commit_lsn,
 723                                                  XLogRecPtr end_lsn)
 724 {
 725         ReorderBufferTXN *txn;
 726         ReorderBufferTXN *subtxn;
 727
 728         subtxn = ReorderBufferTXNByXid(rb, subxid, false, NULL,
 729                                                                    InvalidXLogRecPtr, false);
 730
 731         /*
 732          * No need to do anything if that subtxn didn't contain any changes
 733          */
 734         if (!subtxn)
 735                 return;
 736
 737         txn = ReorderBufferTXNByXid(rb, xid, false, NULL, commit_lsn, true);
 738
 739         if (txn == NULL)
 740                 elog(ERROR, "subxact logged without previous toplevel record");
 741
 742         /*
 743          * Pass the our base snapshot to the parent transaction if it doesn't have
 744          * one, or ours is older. That can happen if there are no changes in the
 745          * toplevel transaction but in one of the child transactions. This allows
 746          * the parent to simply use it's base snapshot initially.
 747          */
 748         if (txn->base_snapshot == NULL ||
 749                 txn->base_snapshot_lsn > subtxn->base_snapshot_lsn)
 750         {
 751                 txn->base_snapshot = subtxn->base_snapshot;
 752                 txn->base_snapshot_lsn = subtxn->base_snapshot_lsn;
 753                 subtxn->base_snapshot = NULL;
 754                 subtxn->base_snapshot_lsn = InvalidXLogRecPtr;
 755         }
 756
 757         subtxn->final_lsn = commit_lsn;
 758         subtxn->end_lsn = end_lsn;
 759
 760         if (!subtxn->is_known_as_subxact)
 761         {
 762                 subtxn->is_known_as_subxact = true;
 763                 Assert(subtxn->nsubtxns == 0);
 764
 765                 /* remove from lsn order list of top-level transactions */
 766                 dlist_delete(&subtxn->node);
 767
 768                 /* add to subtransaction list */
 769                 dlist_push_tail(&txn->subtxns, &subtxn->node);
 770                 txn->nsubtxns++;
 771         }
 772 }
 773
 774
 775 /*
 776  * Support for efficiently iterating over a transaction's and its
 777  * subtransactions' changes.
 778  *
 779  * We do by doing a k-way merge between transactions/subtransactions. For that
 780  * we model the current heads of the different transactions as a binary heap
 781  * so we easily know which (sub-)transaction has the change with the smallest
 782  * lsn next.
 783  *
 784  * We assume the changes in individual transactions are already sorted by LSN.
 785  */
 786
 787 /*
 788  * Binary heap comparison function.
 789  */
 790 static int
 791 ReorderBufferIterCompare(Datum a, Datum b, void *arg)
 792 {
 793         ReorderBufferIterTXNState *state = (ReorderBufferIterTXNState *) arg;
 794         XLogRecPtr      pos_a = state->entries[DatumGetInt32(a)].lsn;
 795         XLogRecPtr      pos_b = state->entries[DatumGetInt32(b)].lsn;
 796
 797         if (pos_a < pos_b)
 798                 return 1;
 799         else if (pos_a == pos_b)
 800                 return 0;
 801         return -1;
 802 }
 803
 804 /*
 805  * Allocate & initialize an iterator which iterates in lsn order over a
 806  * transaction and all its subtransactions.
 807  */
 808 static ReorderBufferIterTXNState *
 809 ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn)
 810 {
 811         Size            nr_txns = 0;
 812         ReorderBufferIterTXNState *state;
 813         dlist_iter      cur_txn_i;
 814         int32           off;
 815
 816         /*
 817          * Calculate the size of our heap: one element for every transaction that
 818          * contains changes.  (Besides the transactions already in the reorder
 819          * buffer, we count the one we were directly passed.)
 820          */
 821         if (txn->nentries > 0)
 822                 nr_txns++;
 823
 824         dlist_foreach(cur_txn_i, &txn->subtxns)
 825         {
 826                 ReorderBufferTXN *cur_txn;
 827
 828                 cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);
 829
 830                 if (cur_txn->nentries > 0)
 831                         nr_txns++;
 832         }
 833
 834         /*
 835          * TODO: Consider adding fastpath for the rather common nr_txns=1 case, no
 836          * need to allocate/build a heap then.
 837          */
 838
 839         /* allocate iteration state */
 840         state = (ReorderBufferIterTXNState *)
 841                 MemoryContextAllocZero(rb->context,
 842                                                            sizeof(ReorderBufferIterTXNState) +
 843                                                            sizeof(ReorderBufferIterTXNEntry) * nr_txns);
 844
 845         state->nr_txns = nr_txns;
 846         dlist_init(&state->old_change);
 847
 848         for (off = 0; off < state->nr_txns; off++)
 849         {
 850                 state->entries[off].fd = -1;
 851                 state->entries[off].segno = 0;
 852         }
 853
 854         /* allocate heap */
 855         state->heap = binaryheap_allocate(state->nr_txns,
 856                                                                           ReorderBufferIterCompare,
 857                                                                           state);
 858
 859         /*
 860          * Now insert items into the binary heap, in an unordered fashion.  (We
 861          * will run a heap assembly step at the end; this is more efficient.)
 862          */
 863
 864         off = 0;
 865
 866         /* add toplevel transaction if it contains changes */
 867         if (txn->nentries > 0)
 868         {
 869                 ReorderBufferChange *cur_change;
 870
 871                 if (txn->nentries != txn->nentries_mem)
 872                         ReorderBufferRestoreChanges(rb, txn, &state->entries[off].fd,
 873                                                                                 &state->entries[off].segno);
 874
 875                 cur_change = dlist_head_element(ReorderBufferChange, node,
 876                                                                                 &txn->changes);
 877
 878                 state->entries[off].lsn = cur_change->lsn;
 879                 state->entries[off].change = cur_change;
 880                 state->entries[off].txn = txn;
 881
 882                 binaryheap_add_unordered(state->heap, Int32GetDatum(off++));
 883         }
 884
 885         /* add subtransactions if they contain changes */
 886         dlist_foreach(cur_txn_i, &txn->subtxns)
 887         {
 888                 ReorderBufferTXN *cur_txn;
 889
 890                 cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);
 891
 892                 if (cur_txn->nentries > 0)
 893                 {
 894                         ReorderBufferChange *cur_change;
 895
 896                         if (txn->nentries != txn->nentries_mem)
 897                                 ReorderBufferRestoreChanges(rb, cur_txn,
 898                                                                                         &state->entries[off].fd,
 899                                                                                         &state->entries[off].segno);
 900
 901                         cur_change = dlist_head_element(ReorderBufferChange, node,
 902                                                                                         &cur_txn->changes);
 903
 904                         state->entries[off].lsn = cur_change->lsn;
 905                         state->entries[off].change = cur_change;
 906                         state->entries[off].txn = cur_txn;
 907
 908                         binaryheap_add_unordered(state->heap, Int32GetDatum(off++));
 909                 }
 910         }
 911
 912         /* assemble a valid binary heap */
 913         binaryheap_build(state->heap);
 914
 915         return state;
 916 }
 917
 918 /*
 919  * Return the next change when iterating over a transaction and its
 920  * subtransactions.
 921  *
 922  * Returns NULL when no further changes exist.
 923  */
 924 static ReorderBufferChange *
 925 ReorderBufferIterTXNNext(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
 926 {
 927         ReorderBufferChange *change;
 928         ReorderBufferIterTXNEntry *entry;
 929         int32           off;
 930
 931         /* nothing there anymore */
 932         if (state->heap->bh_size == 0)
 933                 return NULL;
 934
 935         off = DatumGetInt32(binaryheap_first(state->heap));
 936         entry = &state->entries[off];
 937
 938         /* free memory we might have "leaked" in the previous *Next call */
 939         if (!dlist_is_empty(&state->old_change))
 940         {
 941                 change = dlist_container(ReorderBufferChange, node,
 942                                                                  dlist_pop_head_node(&state->old_change));
 943                 ReorderBufferReturnChange(rb, change);
 944                 Assert(dlist_is_empty(&state->old_change));
 945         }
 946
 947         change = entry->change;
 948
 949         /*
 950          * update heap with information about which transaction has the next
 951          * relevant change in LSN order
 952          */
 953
 954         /* there are in-memory changes */
 955         if (dlist_has_next(&entry->txn->changes, &entry->change->node))
 956         {
 957                 dlist_node *next = dlist_next_node(&entry->txn->changes, &change->node);
 958                 ReorderBufferChange *next_change =
 959                 dlist_container(ReorderBufferChange, node, next);
 960
 961                 /* txn stays the same */
 962                 state->entries[off].lsn = next_change->lsn;
 963                 state->entries[off].change = next_change;
 964
 965                 binaryheap_replace_first(state->heap, Int32GetDatum(off));
 966                 return change;
 967         }
 968
 969         /* try to load changes from disk */
 970         if (entry->txn->nentries != entry->txn->nentries_mem)
 971         {
 972                 /*
 973                  * Ugly: restoring changes will reuse *Change records, thus delete the
 974                  * current one from the per-tx list and only free in the next call.
 975                  */
 976                 dlist_delete(&change->node);
 977                 dlist_push_tail(&state->old_change, &change->node);
 978
 979                 if (ReorderBufferRestoreChanges(rb, entry->txn, &entry->fd,
 980                                                                                 &state->entries[off].segno))
 981                 {
 982                         /* successfully restored changes from disk */
 983                         ReorderBufferChange *next_change =
 984                         dlist_head_element(ReorderBufferChange, node,
 985                                                            &entry->txn->changes);
 986
 987                         elog(DEBUG2, "restored %u/%u changes from disk",
 988                                  (uint32) entry->txn->nentries_mem,
 989                                  (uint32) entry->txn->nentries);
 990
 991                         Assert(entry->txn->nentries_mem);
 992                         /* txn stays the same */
 993                         state->entries[off].lsn = next_change->lsn;
 994                         state->entries[off].change = next_change;
 995                         binaryheap_replace_first(state->heap, Int32GetDatum(off));
 996
 997                         return change;
 998                 }
 999         }
1000
1001         /* ok, no changes there anymore, remove */
1002         binaryheap_remove_first(state->heap);
1003
1004         return change;
1005 }
1006
1007 /*
1008  * Deallocate the iterator
1009  */
1010 static void
1011 ReorderBufferIterTXNFinish(ReorderBuffer *rb,
1012                                                    ReorderBufferIterTXNState *state)
1013 {
1014         int32           off;
1015
1016         for (off = 0; off < state->nr_txns; off++)
1017         {
1018                 if (state->entries[off].fd != -1)
1019                         CloseTransientFile(state->entries[off].fd);
1020         }
1021
1022         /* free memory we might have "leaked" in the last *Next call */
1023         if (!dlist_is_empty(&state->old_change))
1024         {
1025                 ReorderBufferChange *change;
1026
1027                 change = dlist_container(ReorderBufferChange, node,
1028                                                                  dlist_pop_head_node(&state->old_change));
1029                 ReorderBufferReturnChange(rb, change);
1030                 Assert(dlist_is_empty(&state->old_change));
1031         }
1032
1033         binaryheap_free(state->heap);
1034         pfree(state);
1035 }
1036
1037 /*
1038  * Cleanup the contents of a transaction, usually after the transaction
1039  * committed or aborted.
1040  */
1041 static void
1042 ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
1043 {
1044         bool            found;
1045         dlist_mutable_iter iter;
1046
1047         /* cleanup subtransactions & their changes */
1048         dlist_foreach_modify(iter, &txn->subtxns)
1049         {
1050                 ReorderBufferTXN *subtxn;
1051
1052                 subtxn = dlist_container(ReorderBufferTXN, node, iter.cur);
1053
1054                 /*
1055                  * Subtransactions are always associated to the toplevel TXN, even if
1056                  * they originally were happening inside another subtxn, so we won't
1057                  * ever recurse more than one level deep here.
1058                  */
1059                 Assert(subtxn->is_known_as_subxact);
1060                 Assert(subtxn->nsubtxns == 0);
1061
1062                 ReorderBufferCleanupTXN(rb, subtxn);
1063         }
1064
1065         /* cleanup changes in the toplevel txn */
1066         dlist_foreach_modify(iter, &txn->changes)
1067         {
1068                 ReorderBufferChange *change;
1069
1070                 change = dlist_container(ReorderBufferChange, node, iter.cur);
1071
1072                 ReorderBufferReturnChange(rb, change);
1073         }
1074
1075         /*
1076          * Cleanup the tuplecids we stored for decoding catalog snapshot access.
1077          * They are always stored in the toplevel transaction.
1078          */
1079         dlist_foreach_modify(iter, &txn->tuplecids)
1080         {
1081                 ReorderBufferChange *change;
1082
1083                 change = dlist_container(ReorderBufferChange, node, iter.cur);
1084                 Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
1085                 ReorderBufferReturnChange(rb, change);
1086         }
1087
1088         if (txn->base_snapshot != NULL)
1089         {
1090                 SnapBuildSnapDecRefcount(txn->base_snapshot);
1091                 txn->base_snapshot = NULL;
1092                 txn->base_snapshot_lsn = InvalidXLogRecPtr;
1093         }
1094
1095         /* delete from list of known subxacts */
1096         if (txn->is_known_as_subxact)
1097         {
1098                 /* NB: nsubxacts count of parent will be too high now */
1099                 dlist_delete(&txn->node);
1100         }
1101         /* delete from LSN ordered list of toplevel TXNs */
1102         else
1103         {
1104                 dlist_delete(&txn->node);
1105         }
1106
1107         /* now remove reference from buffer */
1108         hash_search(rb->by_txn,
1109                                 (void *) &txn->xid,
1110                                 HASH_REMOVE,
1111                                 &found);
1112         Assert(found);
1113
1114         /* remove entries spilled to disk */
1115         if (txn->nentries != txn->nentries_mem)
1116                 ReorderBufferRestoreCleanup(rb, txn);
1117
1118         /* deallocate */
1119         ReorderBufferReturnTXN(rb, txn);
1120 }
1121
1122 /*
1123  * Build a hash with a (relfilenode, ctid) -> (cmin, cmax) mapping for use by
1124  * tqual.c's HeapTupleSatisfiesHistoricMVCC.
1125  */
1126 static void
1127 ReorderBufferBuildTupleCidHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
1128 {
1129         dlist_iter      iter;
1130         HASHCTL         hash_ctl;
1131
1132         if (!txn->has_catalog_changes || dlist_is_empty(&txn->tuplecids))
1133                 return;
1134
1135         memset(&hash_ctl, 0, sizeof(hash_ctl));
1136
1137         hash_ctl.keysize = sizeof(ReorderBufferTupleCidKey);
1138         hash_ctl.entrysize = sizeof(ReorderBufferTupleCidEnt);
1139         hash_ctl.hcxt = rb->context;
1140
1141         /*
1142          * create the hash with the exact number of to-be-stored tuplecids from
1143          * the start
1144          */
1145         txn->tuplecid_hash =
1146                 hash_create("ReorderBufferTupleCid", txn->ntuplecids, &hash_ctl,
1147                                         HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
1148
1149         dlist_foreach(iter, &txn->tuplecids)
1150         {
1151                 ReorderBufferTupleCidKey key;
1152                 ReorderBufferTupleCidEnt *ent;
1153                 bool            found;
1154                 ReorderBufferChange *change;
1155
1156                 change = dlist_container(ReorderBufferChange, node, iter.cur);
1157
1158                 Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
1159
1160                 /* be careful about padding */
1161                 memset(&key, 0, sizeof(ReorderBufferTupleCidKey));
1162
1163                 key.relnode = change->data.tuplecid.node;
1164
1165                 ItemPointerCopy(&change->data.tuplecid.tid,
1166                                                 &key.tid);
1167
1168                 ent = (ReorderBufferTupleCidEnt *)
1169                         hash_search(txn->tuplecid_hash,
1170                                                 (void *) &key,
1171                                                 HASH_ENTER | HASH_FIND,
1172                                                 &found);
1173                 if (!found)
1174                 {
1175                         ent->cmin = change->data.tuplecid.cmin;
1176                         ent->cmax = change->data.tuplecid.cmax;
1177                         ent->combocid = change->data.tuplecid.combocid;
1178                 }
1179                 else
1180                 {
1181                         Assert(ent->cmin == change->data.tuplecid.cmin);
1182                         Assert(ent->cmax == InvalidCommandId ||
1183                                    ent->cmax == change->data.tuplecid.cmax);
1184
1185                         /*
1186                          * if the tuple got valid in this transaction and now got deleted
1187                          * we already have a valid cmin stored. The cmax will be
1188                          * InvalidCommandId though.
1189                          */
1190                         ent->cmax = change->data.tuplecid.cmax;
1191                 }
1192         }
1193 }
1194
1195 /*
1196  * Copy a provided snapshot so we can modify it privately. This is needed so
1197  * that catalog modifying transactions can look into intermediate catalog
1198  * states.
1199  */
1200 static Snapshot
1201 ReorderBufferCopySnap(ReorderBuffer *rb, Snapshot orig_snap,
1202                                           ReorderBufferTXN *txn, CommandId cid)
1203 {
1204         Snapshot        snap;
1205         dlist_iter      iter;
1206         int                     i = 0;
1207         Size            size;
1208
1209         size = sizeof(SnapshotData) +
1210                 sizeof(TransactionId) * orig_snap->xcnt +
1211                 sizeof(TransactionId) * (txn->nsubtxns + 1);
1212
1213         snap = MemoryContextAllocZero(rb->context, size);
1214         memcpy(snap, orig_snap, sizeof(SnapshotData));
1215
1216         snap->copied = true;
1217         snap->active_count = 1;         /* mark as active so nobody frees it */
1218         snap->regd_count = 0;
1219         snap->xip = (TransactionId *) (snap + 1);
1220
1221         memcpy(snap->xip, orig_snap->xip, sizeof(TransactionId) * snap->xcnt);
1222
1223         /*
1224          * snap->subxip contains all txids that belong to our transaction which we
1225          * need to check via cmin/cmax. Thats why we store the toplevel
1226          * transaction in there as well.
1227          */
1228         snap->subxip = snap->xip + snap->xcnt;
1229         snap->subxip[i++] = txn->xid;
1230
1231         /*
1232          * nsubxcnt isn't decreased when subtransactions abort, so count manually.
1233          * Since it's an upper boundary it is safe to use it for the allocation
1234          * above.
1235          */
1236         snap->subxcnt = 1;
1237
1238         dlist_foreach(iter, &txn->subtxns)
1239         {
1240                 ReorderBufferTXN *sub_txn;
1241
1242                 sub_txn = dlist_container(ReorderBufferTXN, node, iter.cur);
1243                 snap->subxip[i++] = sub_txn->xid;
1244                 snap->subxcnt++;
1245         }
1246
1247         /* sort so we can bsearch() later */
1248         qsort(snap->subxip, snap->subxcnt, sizeof(TransactionId), xidComparator);
1249
1250         /* store the specified current CommandId */
1251         snap->curcid = cid;
1252
1253         return snap;
1254 }
1255
1256 /*
1257  * Free a previously ReorderBufferCopySnap'ed snapshot
1258  */
1259 static void
1260 ReorderBufferFreeSnap(ReorderBuffer *rb, Snapshot snap)
1261 {
1262         if (snap->copied)
1263                 pfree(snap);
1264         else
1265                 SnapBuildSnapDecRefcount(snap);
1266 }
1267
1268 /*
1269  * Perform the replay of a transaction and it's non-aborted subtransactions.
1270  *
1271  * Subtransactions previously have to be processed by
1272  * ReorderBufferCommitChild(), even if previously assigned to the toplevel
1273  * transaction with ReorderBufferAssignChild.
1274  *
1275  * We currently can only decode a transaction's contents in when their commit
1276  * record is read because that's currently the only place where we know about
1277  * cache invalidations. Thus, once a toplevel commit is read, we iterate over
1278  * the top and subtransactions (using a k-way merge) and replay the changes in
1279  * lsn order.
1280  */
1281 void
1282 ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
1283                                         XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
1284                                         TimestampTz commit_time,
1285                                         RepOriginId origin_id, XLogRecPtr origin_lsn)
1286 {
1287         ReorderBufferTXN *txn;
1288         volatile Snapshot snapshot_now;
1289         volatile CommandId command_id = FirstCommandId;
1290         bool            using_subtxn;
1291         ReorderBufferIterTXNState *volatile iterstate = NULL;
1292
1293         txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
1294                                                                 false);
1295
1296         /* unknown transaction, nothing to replay */
1297         if (txn == NULL)
1298                 return;
1299
1300         txn->final_lsn = commit_lsn;
1301         txn->end_lsn = end_lsn;
1302         txn->commit_time = commit_time;
1303         txn->origin_id = origin_id;
1304         txn->origin_lsn = origin_lsn;
1305
1306         /* serialize the last bunch of changes if we need start earlier anyway */
1307         if (txn->nentries_mem != txn->nentries)
1308                 ReorderBufferSerializeTXN(rb, txn);
1309
1310         /*
1311          * If this transaction didn't have any real changes in our database, it's
1312          * OK not to have a snapshot. Note that ReorderBufferCommitChild will have
1313          * transferred its snapshot to this transaction if it had one and the
1314          * toplevel tx didn't.
1315          */
1316         if (txn->base_snapshot == NULL)
1317         {
1318                 Assert(txn->ninvalidations == 0);
1319                 ReorderBufferCleanupTXN(rb, txn);
1320                 return;
1321         }
1322
1323         snapshot_now = txn->base_snapshot;
1324
1325         /* build data to be able to lookup the CommandIds of catalog tuples */
1326         ReorderBufferBuildTupleCidHash(rb, txn);
1327
1328         /* setup the initial snapshot */
1329         SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
1330
1331         /*
1332          * Decoding needs access to syscaches et al., which in turn use
1333          * heavyweight locks and such. Thus we need to have enough state around to
1334          * keep track of those.  The easiest way is to simply use a transaction
1335          * internally.  That also allows us to easily enforce that nothing writes
1336          * to the database by checking for xid assignments.
1337          *
1338          * When we're called via the SQL SRF there's already a transaction
1339          * started, so start an explicit subtransaction there.
1340          */
1341         using_subtxn = IsTransactionOrTransactionBlock();
1342
1343         PG_TRY();
1344         {
1345                 ReorderBufferChange *change;
1346                 ReorderBufferChange *specinsert = NULL;
1347
1348                 if (using_subtxn)
1349                         BeginInternalSubTransaction("replay");
1350                 else
1351                         StartTransactionCommand();
1352
1353                 rb->begin(rb, txn);
1354
1355                 iterstate = ReorderBufferIterTXNInit(rb, txn);
1356                 while ((change = ReorderBufferIterTXNNext(rb, iterstate)) != NULL)
1357                 {
1358                         Relation        relation = NULL;
1359                         Oid                     reloid;
1360
1361                         switch (change->action)
1362                         {
1363                                 case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
1364
1365                                         /*
1366                                          * Confirmation for speculative insertion arrived. Simply
1367                                          * use as a normal record. It'll be cleaned up at the end
1368                                          * of INSERT processing.
1369                                          */
1370                                         Assert(specinsert->data.tp.oldtuple == NULL);
1371                                         change = specinsert;
1372                                         change->action = REORDER_BUFFER_CHANGE_INSERT;
1373
1374                                         /* intentionally fall through */
1375                                 case REORDER_BUFFER_CHANGE_INSERT:
1376                                 case REORDER_BUFFER_CHANGE_UPDATE:
1377                                 case REORDER_BUFFER_CHANGE_DELETE:
1378                                         Assert(snapshot_now);
1379
1380                                         reloid = RelidByRelfilenode(change->data.tp.relnode.spcNode,
1381                                                                                         change->data.tp.relnode.relNode);
1382
1383                                         /*
1384                                          * Catalog tuple without data, emitted while catalog was
1385                                          * in the process of being rewritten.
1386                                          */
1387                                         if (reloid == InvalidOid &&
1388                                                 change->data.tp.newtuple == NULL &&
1389                                                 change->data.tp.oldtuple == NULL)
1390                                                 goto change_done;
1391                                         else if (reloid == InvalidOid)
1392                                                 elog(ERROR, "could not map filenode \"%s\" to relation OID",
1393                                                          relpathperm(change->data.tp.relnode,
1394                                                                                  MAIN_FORKNUM));
1395
1396                                         relation = RelationIdGetRelation(reloid);
1397
1398                                         if (relation == NULL)
1399                                                 elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
1400                                                          reloid,
1401                                                          relpathperm(change->data.tp.relnode,
1402                                                                                  MAIN_FORKNUM));
1403
1404                                         if (!RelationIsLogicallyLogged(relation))
1405                                                 goto change_done;
1406
1407                                         /*
1408                                          * For now ignore sequence changes entirely. Most of the
1409                                          * time they don't log changes using records we
1410                                          * understand, so it doesn't make sense to handle the few
1411                                          * cases we do.
1412                                          */
1413                                         if (relation->rd_rel->relkind == RELKIND_SEQUENCE)
1414                                                 goto change_done;
1415
1416                                         /* user-triggered change */
1417                                         if (!IsToastRelation(relation))
1418                                         {
1419                                                 ReorderBufferToastReplace(rb, txn, relation, change);
1420                                                 rb->apply_change(rb, txn, relation, change);
1421
1422                                                 /*
1423                                                  * Only clear reassembled toast chunks if we're sure
1424                                                  * they're not required anymore. The creator of the
1425                                                  * tuple tells us.
1426                                                  */
1427                                                 if (change->data.tp.clear_toast_afterwards)
1428                                                         ReorderBufferToastReset(rb, txn);
1429                                         }
1430                                         /* we're not interested in toast deletions */
1431                                         else if (change->action == REORDER_BUFFER_CHANGE_INSERT)
1432                                         {
1433                                                 /*
1434                                                  * Need to reassemble the full toasted Datum in
1435                                                  * memory, to ensure the chunks don't get reused till
1436                                                  * we're done remove it from the list of this
1437                                                  * transaction's changes. Otherwise it will get
1438                                                  * freed/reused while restoring spooled data from
1439                                                  * disk.
1440                                                  */
1441                                                 dlist_delete(&change->node);
1442                                                 ReorderBufferToastAppendChunk(rb, txn, relation,
1443                                                                                                           change);
1444                                         }
1445
1446                         change_done:
1447
1448                                         /*
1449                                          * Either speculative insertion was confirmed, or it was
1450                                          * unsuccessful and the record isn't needed anymore.
1451                                          */
1452                                         if (specinsert != NULL)
1453                                         {
1454                                                 ReorderBufferReturnChange(rb, specinsert);
1455                                                 specinsert = NULL;
1456                                         }
1457
1458                                         if (relation != NULL)
1459                                         {
1460                                                 RelationClose(relation);
1461                                                 relation = NULL;
1462                                         }
1463                                         break;
1464
1465                                 case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
1466
1467                                         /*
1468                                          * Speculative insertions are dealt with by delaying the
1469                                          * processing of the insert until the confirmation record
1470                                          * arrives. For that we simply unlink the record from the
1471                                          * chain, so it does not get freed/reused while restoring
1472                                          * spooled data from disk.
1473                                          *
1474                                          * This is safe in the face of concurrent catalog changes
1475                                          * because the relevant relation can't be changed between
1476                                          * speculative insertion and confirmation due to
1477                                          * CheckTableNotInUse() and locking.
1478                                          */
1479
1480                                         /* clear out a pending (and thus failed) speculation */
1481                                         if (specinsert != NULL)
1482                                         {
1483                                                 ReorderBufferReturnChange(rb, specinsert);
1484                                                 specinsert = NULL;
1485                                         }
1486
1487                                         /* and memorize the pending insertion */
1488                                         dlist_delete(&change->node);
1489                                         specinsert = change;
1490                                         break;
1491
1492                                 case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
1493                                         /* get rid of the old */
1494                                         TeardownHistoricSnapshot(false);
1495
1496                                         if (snapshot_now->copied)
1497                                         {
1498                                                 ReorderBufferFreeSnap(rb, snapshot_now);
1499                                                 snapshot_now =
1500                                                         ReorderBufferCopySnap(rb, change->data.snapshot,
1501                                                                                                   txn, command_id);
1502                                         }
1503
1504                                         /*
1505                                          * Restored from disk, need to be careful not to double
1506                                          * free. We could introduce refcounting for that, but for
1507                                          * now this seems infrequent enough not to care.
1508                                          */
1509                                         else if (change->data.snapshot->copied)
1510                                         {
1511                                                 snapshot_now =
1512                                                         ReorderBufferCopySnap(rb, change->data.snapshot,
1513                                                                                                   txn, command_id);
1514                                         }
1515                                         else
1516                                         {
1517                                                 snapshot_now = change->data.snapshot;
1518                                         }
1519
1520
1521                                         /* and continue with the new one */
1522                                         SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
1523                                         break;
1524
1525                                 case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
1526                                         Assert(change->data.command_id != InvalidCommandId);
1527
1528                                         if (command_id < change->data.command_id)
1529                                         {
1530                                                 command_id = change->data.command_id;
1531
1532                                                 if (!snapshot_now->copied)
1533                                                 {
1534                                                         /* we don't use the global one anymore */
1535                                                         snapshot_now = ReorderBufferCopySnap(rb, snapshot_now,
1536                                                                                                                         txn, command_id);
1537                                                 }
1538
1539                                                 snapshot_now->curcid = command_id;
1540
1541                                                 TeardownHistoricSnapshot(false);
1542                                                 SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
1543
1544                                                 /*
1545                                                  * Every time the CommandId is incremented, we could
1546                                                  * see new catalog contents, so execute all
1547                                                  * invalidations.
1548                                                  */
1549                                                 ReorderBufferExecuteInvalidations(rb, txn);
1550                                         }
1551
1552                                         break;
1553
1554                                 case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
1555                                         elog(ERROR, "tuplecid value in changequeue");
1556                                         break;
1557                         }
1558                 }
1559
1560                 /*
1561                  * There's a speculative insertion remaining, just clean in up, it
1562                  * can't have been successful, otherwise we'd gotten a confirmation
1563                  * record.
1564                  */
1565                 if (specinsert)
1566                 {
1567                         ReorderBufferReturnChange(rb, specinsert);
1568                         specinsert = NULL;
1569                 }
1570
1571                 /* clean up the iterator */
1572                 ReorderBufferIterTXNFinish(rb, iterstate);
1573                 iterstate = NULL;
1574
1575                 /* call commit callback */
1576                 rb->commit(rb, txn, commit_lsn);
1577
1578                 /* this is just a sanity check against bad output plugin behaviour */
1579                 if (GetCurrentTransactionIdIfAny() != InvalidTransactionId)
1580                         elog(ERROR, "output plugin used XID %u",
1581                                  GetCurrentTransactionId());
1582
1583                 /* cleanup */
1584                 TeardownHistoricSnapshot(false);
1585
1586                 /*
1587                  * Aborting the current (sub-)transaction as a whole has the right
1588                  * semantics. We want all locks acquired in here to be released, not
1589                  * reassigned to the parent and we do not want any database access
1590                  * have persistent effects.
1591                  */
1592                 AbortCurrentTransaction();
1593
1594                 /* make sure there's no cache pollution */
1595                 ReorderBufferExecuteInvalidations(rb, txn);
1596
1597                 if (using_subtxn)
1598                         RollbackAndReleaseCurrentSubTransaction();
1599
1600                 if (snapshot_now->copied)
1601                         ReorderBufferFreeSnap(rb, snapshot_now);
1602
1603                 /* remove potential on-disk data, and deallocate */
1604                 ReorderBufferCleanupTXN(rb, txn);
1605         }
1606         PG_CATCH();
1607         {
1608                 /* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */
1609                 if (iterstate)
1610                         ReorderBufferIterTXNFinish(rb, iterstate);
1611
1612                 TeardownHistoricSnapshot(true);
1613
1614                 /*
1615                  * Force cache invalidation to happen outside of a valid transaction
1616                  * to prevent catalog access as we just caught an error.
1617                  */
1618                 AbortCurrentTransaction();
1619
1620                 /* make sure there's no cache pollution */
1621                 ReorderBufferExecuteInvalidations(rb, txn);
1622
1623                 if (using_subtxn)
1624                         RollbackAndReleaseCurrentSubTransaction();
1625
1626                 if (snapshot_now->copied)
1627                         ReorderBufferFreeSnap(rb, snapshot_now);
1628
1629                 /* remove potential on-disk data, and deallocate */
1630                 ReorderBufferCleanupTXN(rb, txn);
1631
1632                 PG_RE_THROW();
1633         }
1634         PG_END_TRY();
1635 }
1636
1637 /*
1638  * Abort a transaction that possibly has previous changes. Needs to be first
1639  * called for subtransactions and then for the toplevel xid.
1640  *
1641  * NB: Transactions handled here have to have actively aborted (i.e. have
1642  * produced an abort record). Implicitly aborted transactions are handled via
1643  * ReorderBufferAbortOld(); transactions we're just not interesteded in, but
1644  * which have committed are handled in ReorderBufferForget().
1645  *
1646  * This function purges this transaction and its contents from memory and
1647  * disk.
1648  */
1649 void
1650 ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
1651 {
1652         ReorderBufferTXN *txn;
1653
1654         txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
1655                                                                 false);
1656
1657         /* unknown, nothing to remove */
1658         if (txn == NULL)
1659                 return;
1660
1661         /* cosmetic... */
1662         txn->final_lsn = lsn;
1663
1664         /* remove potential on-disk data, and deallocate */
1665         ReorderBufferCleanupTXN(rb, txn);
1666 }
1667
1668 /*
1669  * Abort all transactions that aren't actually running anymore because the
1670  * server restarted.
1671  *
1672  * NB: These really have to be transactions that have aborted due to a server
1673  * crash/immediate restart, as we don't deal with invalidations here.
1674  */
1675 void
1676 ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
1677 {
1678         dlist_mutable_iter it;
1679
1680         /*
1681          * Iterate through all (potential) toplevel TXNs and abort all that are
1682          * older than what possibly can be running. Once we've found the first
1683          * that is alive we stop, there might be some that acquired an xid earlier
1684          * but started writing later, but it's unlikely and they will cleaned up
1685          * in a later call to ReorderBufferAbortOld().
1686          */
1687         dlist_foreach_modify(it, &rb->toplevel_by_lsn)
1688         {
1689                 ReorderBufferTXN *txn;
1690
1691                 txn = dlist_container(ReorderBufferTXN, node, it.cur);
1692
1693                 if (TransactionIdPrecedes(txn->xid, oldestRunningXid))
1694                 {
1695                         elog(DEBUG1, "aborting old transaction %u", txn->xid);
1696
1697                         /* remove potential on-disk data, and deallocate this tx */
1698                         ReorderBufferCleanupTXN(rb, txn);
1699                 }
1700                 else
1701                         return;
1702         }
1703 }
1704
1705 /*
1706  * Forget the contents of a transaction if we aren't interested in it's
1707  * contents. Needs to be first called for subtransactions and then for the
1708  * toplevel xid.
1709  *
1710  * This is significantly different to ReorderBufferAbort() because
1711  * transactions that have committed need to be treated differenly from aborted
1712  * ones since they may have modified the catalog.
1713  *
1714  * Note that this is only allowed to be called in the moment a transaction
1715  * commit has just been read, not earlier; otherwise later records referring
1716  * to this xid might re-create the transaction incompletely.
1717  */
1718 void
1719 ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
1720 {
1721         ReorderBufferTXN *txn;
1722
1723         txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
1724                                                                 false);
1725
1726         /* unknown, nothing to forget */
1727         if (txn == NULL)
1728                 return;
1729
1730         /* cosmetic... */
1731         txn->final_lsn = lsn;
1732
1733         /*
1734          * Process cache invalidation messages if there are any. Even if we're not
1735          * interested in the transaction's contents, it could have manipulated the
1736          * catalog and we need to update the caches according to that.
1737          */
1738         if (txn->base_snapshot != NULL && txn->ninvalidations > 0)
1739         {
1740                 bool            use_subtxn = IsTransactionOrTransactionBlock();
1741
1742                 if (use_subtxn)
1743                         BeginInternalSubTransaction("replay");
1744
1745                 /*
1746                  * Force invalidations to happen outside of a valid transaction - that
1747                  * way entries will just be marked as invalid without accessing the
1748                  * catalog. That's advantageous because we don't need to setup the
1749                  * full state necessary for catalog access.
1750                  */
1751                 if (use_subtxn)
1752                         AbortCurrentTransaction();
1753
1754                 ReorderBufferExecuteInvalidations(rb, txn);
1755
1756                 if (use_subtxn)
1757                         RollbackAndReleaseCurrentSubTransaction();
1758         }
1759         else
1760                 Assert(txn->ninvalidations == 0);
1761
1762         /* remove potential on-disk data, and deallocate */
1763         ReorderBufferCleanupTXN(rb, txn);
1764 }
1765
1766
1767 /*
1768  * Tell reorderbuffer about an xid seen in the WAL stream. Has to be called at
1769  * least once for every xid in XLogRecord->xl_xid (other places in records
1770  * may, but do not have to be passed through here).
1771  *
1772  * Reorderbuffer keeps some datastructures about transactions in LSN order,
1773  * for efficiency. To do that it has to know about when transactions are seen
1774  * first in the WAL. As many types of records are not actually interesting for
1775  * logical decoding, they do not necessarily pass though here.
1776  */
1777 void
1778 ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
1779 {
1780         /* many records won't have an xid assigned, centralize check here */
1781         if (xid != InvalidTransactionId)
1782                 ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
1783 }
1784
1785 /*
1786  * Add a new snapshot to this transaction that may only used after lsn 'lsn'
1787  * because the previous snapshot doesn't describe the catalog correctly for
1788  * following rows.
1789  */
1790 void
1791 ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid,
1792                                                  XLogRecPtr lsn, Snapshot snap)
1793 {
1794         ReorderBufferChange *change = ReorderBufferGetChange(rb);
1795
1796         change->data.snapshot = snap;
1797         change->action = REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT;
1798
1799         ReorderBufferQueueChange(rb, xid, lsn, change);
1800 }
1801
1802 /*
1803  * Setup the base snapshot of a transaction. The base snapshot is the snapshot
1804  * that is used to decode all changes until either this transaction modifies
1805  * the catalog or another catalog modifying transaction commits.
1806  *
1807  * Needs to be called before any changes are added with
1808  * ReorderBufferQueueChange().
1809  */
1810 void
1811 ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid,
1812                                                          XLogRecPtr lsn, Snapshot snap)
1813 {
1814         ReorderBufferTXN *txn;
1815         bool            is_new;
1816
1817         txn = ReorderBufferTXNByXid(rb, xid, true, &is_new, lsn, true);
1818         Assert(txn->base_snapshot == NULL);
1819         Assert(snap != NULL);
1820
1821         txn->base_snapshot = snap;
1822         txn->base_snapshot_lsn = lsn;
1823 }
1824
1825 /*
1826  * Access the catalog with this CommandId at this point in the changestream.
1827  *
1828  * May only be called for command ids > 1
1829  */
1830 void
1831 ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid,
1832                                                          XLogRecPtr lsn, CommandId cid)
1833 {
1834         ReorderBufferChange *change = ReorderBufferGetChange(rb);
1835
1836         change->data.command_id = cid;
1837         change->action = REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID;
1838
1839         ReorderBufferQueueChange(rb, xid, lsn, change);
1840 }
1841
1842
1843 /*
1844  * Add new (relfilenode, tid) -> (cmin, cmax) mappings.
1845  */
1846 void
1847 ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid,
1848                                                          XLogRecPtr lsn, RelFileNode node,
1849                                                          ItemPointerData tid, CommandId cmin,
1850                                                          CommandId cmax, CommandId combocid)
1851 {
1852         ReorderBufferChange *change = ReorderBufferGetChange(rb);
1853         ReorderBufferTXN *txn;
1854
1855         txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
1856
1857         change->data.tuplecid.node = node;
1858         change->data.tuplecid.tid = tid;
1859         change->data.tuplecid.cmin = cmin;
1860         change->data.tuplecid.cmax = cmax;
1861         change->data.tuplecid.combocid = combocid;
1862         change->lsn = lsn;
1863         change->action = REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID;
1864
1865         dlist_push_tail(&txn->tuplecids, &change->node);
1866         txn->ntuplecids++;
1867 }
1868
1869 /*
1870  * Setup the invalidation of the toplevel transaction.
1871  *
1872  * This needs to be done before ReorderBufferCommit is called!
1873  */
1874 void
1875 ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid,
1876                                                           XLogRecPtr lsn, Size nmsgs,
1877                                                           SharedInvalidationMessage *msgs)
1878 {
1879         ReorderBufferTXN *txn;
1880
1881         txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
1882
1883         if (txn->ninvalidations != 0)
1884                 elog(ERROR, "only ever add one set of invalidations");
1885
1886         Assert(nmsgs > 0);
1887
1888         txn->ninvalidations = nmsgs;
1889         txn->invalidations = (SharedInvalidationMessage *)
1890                 MemoryContextAlloc(rb->context,
1891                                                    sizeof(SharedInvalidationMessage) * nmsgs);
1892         memcpy(txn->invalidations, msgs,
1893                    sizeof(SharedInvalidationMessage) * nmsgs);
1894 }
1895
1896 /*
1897  * Apply all invalidations we know. Possibly we only need parts at this point
1898  * in the changestream but we don't know which those are.
1899  */
1900 static void
1901 ReorderBufferExecuteInvalidations(ReorderBuffer *rb, ReorderBufferTXN *txn)
1902 {
1903         int                     i;
1904
1905         for (i = 0; i < txn->ninvalidations; i++)
1906                 LocalExecuteInvalidationMessage(&txn->invalidations[i]);
1907 }
1908
1909 /*
1910  * Mark a transaction as containing catalog changes
1911  */
1912 void
1913 ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid,
1914                                                                   XLogRecPtr lsn)
1915 {
1916         ReorderBufferTXN *txn;
1917
1918         txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
1919
1920         txn->has_catalog_changes = true;
1921 }
1922
1923 /*
1924  * Query whether a transaction is already *known* to contain catalog
1925  * changes. This can be wrong until directly before the commit!
1926  */
1927 bool
1928 ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
1929 {
1930         ReorderBufferTXN *txn;
1931
1932         txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
1933                                                                 false);
1934         if (txn == NULL)
1935                 return false;
1936
1937         return txn->has_catalog_changes;
1938 }
1939
1940 /*
1941  * Have we already added the first snapshot?
1942  */
1943 bool
1944 ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
1945 {
1946         ReorderBufferTXN *txn;
1947
1948         txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
1949                                                                 false);
1950
1951         /* transaction isn't known yet, ergo no snapshot */
1952         if (txn == NULL)
1953                 return false;
1954
1955         /*
1956          * TODO: It would be a nice improvement if we would check the toplevel
1957          * transaction in subtransactions, but we'd need to keep track of a bit
1958          * more state.
1959          */
1960         return txn->base_snapshot != NULL;
1961 }
1962
1963
1964 /*
1965  * ---------------------------------------
1966  * Disk serialization support
1967  * ---------------------------------------
1968  */
1969
1970 /*
1971  * Ensure the IO buffer is >= sz.
1972  */
1973 static void
1974 ReorderBufferSerializeReserve(ReorderBuffer *rb, Size sz)
1975 {
1976         if (!rb->outbufsize)
1977         {
1978                 rb->outbuf = MemoryContextAlloc(rb->context, sz);
1979                 rb->outbufsize = sz;
1980         }
1981         else if (rb->outbufsize < sz)
1982         {
1983                 rb->outbuf = repalloc(rb->outbuf, sz);
1984                 rb->outbufsize = sz;
1985         }
1986 }
1987
1988 /*
1989  * Check whether the transaction tx should spill its data to disk.
1990  */
1991 static void
1992 ReorderBufferCheckSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
1993 {
1994         /*
1995          * TODO: improve accounting so we cheaply can take subtransactions into
1996          * account here.
1997          */
1998         if (txn->nentries_mem >= max_changes_in_memory)
1999         {
2000                 ReorderBufferSerializeTXN(rb, txn);
2001                 Assert(txn->nentries_mem == 0);
2002         }
2003 }
2004
2005 /*
2006  * Spill data of a large transaction (and its subtransactions) to disk.
2007  */
2008 static void
2009 ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
2010 {
2011         dlist_iter      subtxn_i;
2012         dlist_mutable_iter change_i;
2013         int                     fd = -1;
2014         XLogSegNo       curOpenSegNo = 0;
2015         Size            spilled = 0;
2016         char            path[MAXPGPATH];
2017
2018         elog(DEBUG2, "spill %u changes in XID %u to disk",
2019                  (uint32) txn->nentries_mem, txn->xid);
2020
2021         /* do the same to all child TXs */
2022         dlist_foreach(subtxn_i, &txn->subtxns)
2023         {
2024                 ReorderBufferTXN *subtxn;
2025
2026                 subtxn = dlist_container(ReorderBufferTXN, node, subtxn_i.cur);
2027                 ReorderBufferSerializeTXN(rb, subtxn);
2028         }
2029
2030         /* serialize changestream */
2031         dlist_foreach_modify(change_i, &txn->changes)
2032         {
2033                 ReorderBufferChange *change;
2034
2035                 change = dlist_container(ReorderBufferChange, node, change_i.cur);
2036
2037                 /*
2038                  * store in segment in which it belongs by start lsn, don't split over
2039                  * multiple segments tho
2040                  */
2041                 if (fd == -1 || !XLByteInSeg(change->lsn, curOpenSegNo))
2042                 {
2043                         XLogRecPtr      recptr;
2044
2045                         if (fd != -1)
2046                                 CloseTransientFile(fd);
2047
2048                         XLByteToSeg(change->lsn, curOpenSegNo);
2049                         XLogSegNoOffsetToRecPtr(curOpenSegNo, 0, recptr);
2050
2051                         /*
2052                          * No need to care about TLIs here, only used during a single run,
2053                          * so each LSN only maps to a specific WAL record.
2054                          */
2055                         sprintf(path, "pg_replslot/%s/xid-%u-lsn-%X-%X.snap",
2056                                         NameStr(MyReplicationSlot->data.name), txn->xid,
2057                                         (uint32) (recptr >> 32), (uint32) recptr);
2058
2059                         /* open segment, create it if necessary */
2060                         fd = OpenTransientFile(path,
2061                                                                    O_CREAT | O_WRONLY | O_APPEND | PG_BINARY,
2062                                                                    S_IRUSR | S_IWUSR);
2063
2064                         if (fd < 0)
2065                                 ereport(ERROR,
2066                                                 (errcode_for_file_access(),
2067                                                  errmsg("could not open file \"%s\": %m",
2068                                                                 path)));
2069                 }
2070
2071                 ReorderBufferSerializeChange(rb, txn, fd, change);
2072                 dlist_delete(&change->node);
2073                 ReorderBufferReturnChange(rb, change);
2074
2075                 spilled++;
2076         }
2077
2078         Assert(spilled == txn->nentries_mem);
2079         Assert(dlist_is_empty(&txn->changes));
2080         txn->nentries_mem = 0;
2081
2082         if (fd != -1)
2083                 CloseTransientFile(fd);
2084 }
2085
2086 /*
2087  * Serialize individual change to disk.
2088  */
2089 static void
2090 ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
2091                                                          int fd, ReorderBufferChange *change)
2092 {
2093         ReorderBufferDiskChange *ondisk;
2094         Size            sz = sizeof(ReorderBufferDiskChange);
2095
2096         ReorderBufferSerializeReserve(rb, sz);
2097
2098         ondisk = (ReorderBufferDiskChange *) rb->outbuf;
2099         memcpy(&ondisk->change, change, sizeof(ReorderBufferChange));
2100
2101         switch (change->action)
2102         {
2103                         /* fall through these, they're all similar enough */
2104                 case REORDER_BUFFER_CHANGE_INSERT:
2105                 case REORDER_BUFFER_CHANGE_UPDATE:
2106                 case REORDER_BUFFER_CHANGE_DELETE:
2107                 case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
2108                         {
2109                                 char       *data;
2110                                 ReorderBufferTupleBuf *oldtup,
2111                                                    *newtup;
2112                                 Size            oldlen = 0;
2113                                 Size            newlen = 0;
2114
2115                                 oldtup = change->data.tp.oldtuple;
2116                                 newtup = change->data.tp.newtuple;
2117
2118                                 if (oldtup)
2119                                 {
2120                                         sz += sizeof(HeapTupleData);
2121                                         oldlen = oldtup->tuple.t_len;
2122                                         sz += oldlen;
2123                                 }
2124
2125                                 if (newtup)
2126                                 {
2127                                         sz += sizeof(HeapTupleData);
2128                                         newlen = newtup->tuple.t_len;
2129                                         sz += newlen;
2130                                 }
2131
2132                                 /* make sure we have enough space */
2133                                 ReorderBufferSerializeReserve(rb, sz);
2134
2135                                 data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
2136                                 /* might have been reallocated above */
2137                                 ondisk = (ReorderBufferDiskChange *) rb->outbuf;
2138
2139                                 if (oldlen)
2140                                 {
2141                                         memcpy(data, &oldtup->tuple, sizeof(HeapTupleData));
2142                                         data += sizeof(HeapTupleData);
2143
2144                                         memcpy(data, oldtup->tuple.t_data, oldlen);
2145                                         data += oldlen;
2146                                 }
2147
2148                                 if (newlen)
2149                                 {
2150                                         memcpy(data, &newtup->tuple, sizeof(HeapTupleData));
2151                                         data += sizeof(HeapTupleData);
2152
2153                                         memcpy(data, newtup->tuple.t_data, newlen);
2154                                         data += oldlen;
2155                                 }
2156                                 break;
2157                         }
2158                 case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
2159                         {
2160                                 Snapshot        snap;
2161                                 char       *data;
2162
2163                                 snap = change->data.snapshot;
2164
2165                                 sz += sizeof(SnapshotData) +
2166                                         sizeof(TransactionId) * snap->xcnt +
2167                                         sizeof(TransactionId) * snap->subxcnt
2168                                         ;
2169
2170                                 /* make sure we have enough space */
2171                                 ReorderBufferSerializeReserve(rb, sz);
2172                                 data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
2173                                 /* might have been reallocated above */
2174                                 ondisk = (ReorderBufferDiskChange *) rb->outbuf;
2175
2176                                 memcpy(data, snap, sizeof(SnapshotData));
2177                                 data += sizeof(SnapshotData);
2178
2179                                 if (snap->xcnt)
2180                                 {
2181                                         memcpy(data, snap->xip,
2182                                                    sizeof(TransactionId) * snap->xcnt);
2183                                         data += sizeof(TransactionId) * snap->xcnt;
2184                                 }
2185
2186                                 if (snap->subxcnt)
2187                                 {
2188                                         memcpy(data, snap->subxip,
2189                                                    sizeof(TransactionId) * snap->subxcnt);
2190                                         data += sizeof(TransactionId) * snap->subxcnt;
2191                                 }
2192                                 break;
2193                         }
2194                 case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
2195                 case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
2196                 case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
2197                         /* ReorderBufferChange contains everything important */
2198                         break;
2199         }
2200
2201         ondisk->size = sz;
2202
2203         if (write(fd, rb->outbuf, ondisk->size) != ondisk->size)
2204         {
2205                 CloseTransientFile(fd);
2206                 ereport(ERROR,
2207                                 (errcode_for_file_access(),
2208                                  errmsg("could not write to data file for XID %u: %m",
2209                                                 txn->xid)));
2210         }
2211
2212         Assert(ondisk->change.action == change->action);
2213 }
2214
2215 /*
2216  * Restore a number of changes spilled to disk back into memory.
2217  */
2218 static Size
2219 ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
2220                                                         int *fd, XLogSegNo *segno)
2221 {
2222         Size            restored = 0;
2223         XLogSegNo       last_segno;
2224         dlist_mutable_iter cleanup_iter;
2225
2226         Assert(txn->first_lsn != InvalidXLogRecPtr);
2227         Assert(txn->final_lsn != InvalidXLogRecPtr);
2228
2229         /* free current entries, so we have memory for more */
2230         dlist_foreach_modify(cleanup_iter, &txn->changes)
2231         {
2232                 ReorderBufferChange *cleanup =
2233                 dlist_container(ReorderBufferChange, node, cleanup_iter.cur);
2234
2235                 dlist_delete(&cleanup->node);
2236                 ReorderBufferReturnChange(rb, cleanup);
2237         }
2238         txn->nentries_mem = 0;
2239         Assert(dlist_is_empty(&txn->changes));
2240
2241         XLByteToSeg(txn->final_lsn, last_segno);
2242
2243         while (restored < max_changes_in_memory && *segno <= last_segno)
2244         {
2245                 int                     readBytes;
2246                 ReorderBufferDiskChange *ondisk;
2247
2248                 if (*fd == -1)
2249                 {
2250                         XLogRecPtr      recptr;
2251                         char            path[MAXPGPATH];
2252
2253                         /* first time in */
2254                         if (*segno == 0)
2255                         {
2256                                 XLByteToSeg(txn->first_lsn, *segno);
2257                         }
2258
2259                         Assert(*segno != 0 || dlist_is_empty(&txn->changes));
2260                         XLogSegNoOffsetToRecPtr(*segno, 0, recptr);
2261
2262                         /*
2263                          * No need to care about TLIs here, only used during a single run,
2264                          * so each LSN only maps to a specific WAL record.
2265                          */
2266                         sprintf(path, "pg_replslot/%s/xid-%u-lsn-%X-%X.snap",
2267                                         NameStr(MyReplicationSlot->data.name), txn->xid,
2268                                         (uint32) (recptr >> 32), (uint32) recptr);
2269
2270                         *fd = OpenTransientFile(path, O_RDONLY | PG_BINARY, 0);
2271                         if (*fd < 0 && errno == ENOENT)
2272                         {
2273                                 *fd = -1;
2274                                 (*segno)++;
2275                                 continue;
2276                         }
2277                         else if (*fd < 0)
2278                                 ereport(ERROR,
2279                                                 (errcode_for_file_access(),
2280                                                  errmsg("could not open file \"%s\": %m",
2281                                                                 path)));
2282
2283                 }
2284
2285                 /*
2286                  * Read the statically sized part of a change which has information
2287                  * about the total size. If we couldn't read a record, we're at the
2288                  * end of this file.
2289                  */
2290                 ReorderBufferSerializeReserve(rb, sizeof(ReorderBufferDiskChange));
2291                 readBytes = read(*fd, rb->outbuf, sizeof(ReorderBufferDiskChange));
2292
2293                 /* eof */
2294                 if (readBytes == 0)
2295                 {
2296                         CloseTransientFile(*fd);
2297                         *fd = -1;
2298                         (*segno)++;
2299                         continue;
2300                 }
2301                 else if (readBytes < 0)
2302                         ereport(ERROR,
2303                                         (errcode_for_file_access(),
2304                                 errmsg("could not read from reorderbuffer spill file: %m")));
2305                 else if (readBytes != sizeof(ReorderBufferDiskChange))
2306                         ereport(ERROR,
2307                                         (errcode_for_file_access(),
2308                                          errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
2309                                                         readBytes,
2310                                                         (uint32) sizeof(ReorderBufferDiskChange))));
2311
2312                 ondisk = (ReorderBufferDiskChange *) rb->outbuf;
2313
2314                 ReorderBufferSerializeReserve(rb,
2315                                                          sizeof(ReorderBufferDiskChange) + ondisk->size);
2316                 ondisk = (ReorderBufferDiskChange *) rb->outbuf;
2317
2318                 readBytes = read(*fd, rb->outbuf + sizeof(ReorderBufferDiskChange),
2319                                                  ondisk->size - sizeof(ReorderBufferDiskChange));
2320
2321                 if (readBytes < 0)
2322                         ereport(ERROR,
2323                                         (errcode_for_file_access(),
2324                                 errmsg("could not read from reorderbuffer spill file: %m")));
2325                 else if (readBytes != ondisk->size - sizeof(ReorderBufferDiskChange))
2326                         ereport(ERROR,
2327                                         (errcode_for_file_access(),
2328                                          errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
2329                                                         readBytes,
2330                                 (uint32) (ondisk->size - sizeof(ReorderBufferDiskChange)))));
2331
2332                 /*
2333                  * ok, read a full change from disk, now restore it into proper
2334                  * in-memory format
2335                  */
2336                 ReorderBufferRestoreChange(rb, txn, rb->outbuf);
2337                 restored++;
2338         }
2339
2340         return restored;
2341 }
2342
2343 /*
2344  * Convert change from its on-disk format to in-memory format and queue it onto
2345  * the TXN's ->changes list.
2346  */
2347 static void
2348 ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
2349                                                    char *data)
2350 {
2351         ReorderBufferDiskChange *ondisk;
2352         ReorderBufferChange *change;
2353
2354         ondisk = (ReorderBufferDiskChange *) data;
2355
2356         change = ReorderBufferGetChange(rb);
2357
2358         /* copy static part */
2359         memcpy(change, &ondisk->change, sizeof(ReorderBufferChange));
2360
2361         data += sizeof(ReorderBufferDiskChange);
2362
2363         /* restore individual stuff */
2364         switch (change->action)
2365         {
2366                         /* fall through these, they're all similar enough */
2367                 case REORDER_BUFFER_CHANGE_INSERT:
2368                 case REORDER_BUFFER_CHANGE_UPDATE:
2369                 case REORDER_BUFFER_CHANGE_DELETE:
2370                 case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
2371                         if (change->data.tp.oldtuple)
2372                         {
2373                                 Size            tuplelen = ((HeapTuple) data)->t_len;
2374
2375                                 change->data.tp.oldtuple =
2376                                         ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader);
2377
2378                                 /* restore ->tuple */
2379                                 memcpy(&change->data.tp.oldtuple->tuple, data,
2380                                            sizeof(HeapTupleData));
2381                                 data += sizeof(HeapTupleData);
2382
2383                                 /* reset t_data pointer into the new tuplebuf */
2384                                 change->data.tp.oldtuple->tuple.t_data =
2385                                         ReorderBufferTupleBufData(change->data.tp.oldtuple);
2386
2387                                 /* restore tuple data itself */
2388                                 memcpy(change->data.tp.oldtuple->tuple.t_data, data, tuplelen);
2389                                 data += tuplelen;
2390                         }
2391
2392                         if (change->data.tp.newtuple)
2393                         {
2394                                 Size            tuplelen = ((HeapTuple) data)->t_len;
2395
2396                                 change->data.tp.newtuple =
2397                                         ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader);
2398
2399                                 /* restore ->tuple */
2400                                 memcpy(&change->data.tp.newtuple->tuple, data,
2401                                            sizeof(HeapTupleData));
2402                                 data += sizeof(HeapTupleData);
2403
2404                                 /* reset t_data pointer into the new tuplebuf */
2405                                 change->data.tp.newtuple->tuple.t_data =
2406                                         ReorderBufferTupleBufData(change->data.tp.newtuple);
2407
2408                                 /* restore tuple data itself */
2409                                 memcpy(change->data.tp.newtuple->tuple.t_data, data, tuplelen);
2410                                 data += tuplelen;
2411                         }
2412
2413                         break;
2414                 case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
2415                         {
2416                                 Snapshot        oldsnap;
2417                                 Snapshot        newsnap;
2418                                 Size            size;
2419
2420                                 oldsnap = (Snapshot) data;
2421
2422                                 size = sizeof(SnapshotData) +
2423                                         sizeof(TransactionId) * oldsnap->xcnt +
2424                                         sizeof(TransactionId) * (oldsnap->subxcnt + 0);
2425
2426                                 change->data.snapshot = MemoryContextAllocZero(rb->context, size);
2427
2428                                 newsnap = change->data.snapshot;
2429
2430                                 memcpy(newsnap, data, size);
2431                                 newsnap->xip = (TransactionId *)
2432                                         (((char *) newsnap) + sizeof(SnapshotData));
2433                                 newsnap->subxip = newsnap->xip + newsnap->xcnt;
2434                                 newsnap->copied = true;
2435                                 break;
2436                         }
2437                         /* the base struct contains all the data, easy peasy */
2438                 case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
2439                 case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
2440                 case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
2441                         break;
2442         }
2443
2444         dlist_push_tail(&txn->changes, &change->node);
2445         txn->nentries_mem++;
2446 }
2447
2448 /*
2449  * Remove all on-disk stored for the passed in transaction.
2450  */
2451 static void
2452 ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn)
2453 {
2454         XLogSegNo       first;
2455         XLogSegNo       cur;
2456         XLogSegNo       last;
2457
2458         Assert(txn->first_lsn != InvalidXLogRecPtr);
2459         Assert(txn->final_lsn != InvalidXLogRecPtr);
2460
2461         XLByteToSeg(txn->first_lsn, first);
2462         XLByteToSeg(txn->final_lsn, last);
2463
2464         /* iterate over all possible filenames, and delete them */
2465         for (cur = first; cur <= last; cur++)
2466         {
2467                 char            path[MAXPGPATH];
2468                 XLogRecPtr      recptr;
2469
2470                 XLogSegNoOffsetToRecPtr(cur, 0, recptr);
2471
2472                 sprintf(path, "pg_replslot/%s/xid-%u-lsn-%X-%X.snap",
2473                                 NameStr(MyReplicationSlot->data.name), txn->xid,
2474                                 (uint32) (recptr >> 32), (uint32) recptr);
2475                 if (unlink(path) != 0 && errno != ENOENT)
2476                         ereport(ERROR,
2477                                         (errcode_for_file_access(),
2478                                          errmsg("could not remove file \"%s\": %m", path)));
2479         }
2480 }
2481
2482 /*
2483  * Delete all data spilled to disk after we've restarted/crashed. It will be
2484  * recreated when the respective slots are reused.
2485  */
2486 void
2487 StartupReorderBuffer(void)
2488 {
2489         DIR                *logical_dir;
2490         struct dirent *logical_de;
2491
2492         DIR                *spill_dir;
2493         struct dirent *spill_de;
2494
2495         logical_dir = AllocateDir("pg_replslot");
2496         while ((logical_de = ReadDir(logical_dir, "pg_replslot")) != NULL)
2497         {
2498                 struct stat statbuf;
2499                 char            path[MAXPGPATH];
2500
2501                 if (strcmp(logical_de->d_name, ".") == 0 ||
2502                         strcmp(logical_de->d_name, "..") == 0)
2503                         continue;
2504
2505                 /* if it cannot be a slot, skip the directory */
2506                 if (!ReplicationSlotValidateName(logical_de->d_name, DEBUG2))
2507                         continue;
2508
2509                 /*
2510                  * ok, has to be a surviving logical slot, iterate and delete
2511                  * everythign starting with xid-*
2512                  */
2513                 sprintf(path, "pg_replslot/%s", logical_de->d_name);
2514
2515                 /* we're only creating directories here, skip if it's not our's */
2516                 if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
2517                         continue;
2518
2519                 spill_dir = AllocateDir(path);
2520                 while ((spill_de = ReadDir(spill_dir, path)) != NULL)
2521                 {
2522                         if (strcmp(spill_de->d_name, ".") == 0 ||
2523                                 strcmp(spill_de->d_name, "..") == 0)
2524                                 continue;
2525
2526                         /* only look at names that can be ours */
2527                         if (strncmp(spill_de->d_name, "xid", 3) == 0)
2528                         {
2529                                 sprintf(path, "pg_replslot/%s/%s", logical_de->d_name,
2530                                                 spill_de->d_name);
2531
2532                                 if (unlink(path) != 0)
2533                                         ereport(PANIC,
2534                                                         (errcode_for_file_access(),
2535                                                          errmsg("could not remove file \"%s\": %m",
2536                                                                         path)));
2537                         }
2538                 }
2539                 FreeDir(spill_dir);
2540         }
2541         FreeDir(logical_dir);
2542 }
2543
2544 /* ---------------------------------------
2545  * toast reassembly support
2546  * ---------------------------------------
2547  */
2548
2549 /*
2550  * Initialize per tuple toast reconstruction support.
2551  */
2552 static void
2553 ReorderBufferToastInitHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
2554 {
2555         HASHCTL         hash_ctl;
2556
2557         Assert(txn->toast_hash == NULL);
2558
2559         memset(&hash_ctl, 0, sizeof(hash_ctl));
2560         hash_ctl.keysize = sizeof(Oid);
2561         hash_ctl.entrysize = sizeof(ReorderBufferToastEnt);
2562         hash_ctl.hcxt = rb->context;
2563         txn->toast_hash = hash_create("ReorderBufferToastHash", 5, &hash_ctl,
2564                                                                   HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
2565 }
2566
2567 /*
2568  * Per toast-chunk handling for toast reconstruction
2569  *
2570  * Appends a toast chunk so we can reconstruct it when the tuple "owning" the
2571  * toasted Datum comes along.
2572  */
2573 static void
2574 ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
2575                                                           Relation relation, ReorderBufferChange *change)
2576 {
2577         ReorderBufferToastEnt *ent;
2578         ReorderBufferTupleBuf *newtup;
2579         bool            found;
2580         int32           chunksize;
2581         bool            isnull;
2582         Pointer         chunk;
2583         TupleDesc       desc = RelationGetDescr(relation);
2584         Oid                     chunk_id;
2585         int32           chunk_seq;
2586
2587         if (txn->toast_hash == NULL)
2588                 ReorderBufferToastInitHash(rb, txn);
2589
2590         Assert(IsToastRelation(relation));
2591
2592         newtup = change->data.tp.newtuple;
2593         chunk_id = DatumGetObjectId(fastgetattr(&newtup->tuple, 1, desc, &isnull));
2594         Assert(!isnull);
2595         chunk_seq = DatumGetInt32(fastgetattr(&newtup->tuple, 2, desc, &isnull));
2596         Assert(!isnull);
2597
2598         ent = (ReorderBufferToastEnt *)
2599                 hash_search(txn->toast_hash,
2600                                         (void *) &chunk_id,
2601                                         HASH_ENTER,
2602                                         &found);
2603
2604         if (!found)
2605         {
2606                 Assert(ent->chunk_id == chunk_id);
2607                 ent->num_chunks = 0;
2608                 ent->last_chunk_seq = 0;
2609                 ent->size = 0;
2610                 ent->reconstructed = NULL;
2611                 dlist_init(&ent->chunks);
2612
2613                 if (chunk_seq != 0)
2614                         elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq 0",
2615                                  chunk_seq, chunk_id);
2616         }
2617         else if (found && chunk_seq != ent->last_chunk_seq + 1)
2618                 elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq %d",
2619                          chunk_seq, chunk_id, ent->last_chunk_seq + 1);
2620
2621         chunk = DatumGetPointer(fastgetattr(&newtup->tuple, 3, desc, &isnull));
2622         Assert(!isnull);
2623
2624         /* calculate size so we can allocate the right size at once later */
2625         if (!VARATT_IS_EXTENDED(chunk))
2626                 chunksize = VARSIZE(chunk) - VARHDRSZ;
2627         else if (VARATT_IS_SHORT(chunk))
2628                 /* could happen due to heap_form_tuple doing its thing */
2629                 chunksize = VARSIZE_SHORT(chunk) - VARHDRSZ_SHORT;
2630         else
2631                 elog(ERROR, "unexpected type of toast chunk");
2632
2633         ent->size += chunksize;
2634         ent->last_chunk_seq = chunk_seq;
2635         ent->num_chunks++;
2636         dlist_push_tail(&ent->chunks, &change->node);
2637 }
2638
2639 /*
2640  * Rejigger change->newtuple to point to in-memory toast tuples instead to
2641  * on-disk toast tuples that may not longer exist (think DROP TABLE or VACUUM).
2642  *
2643  * We cannot replace unchanged toast tuples though, so those will still point
2644  * to on-disk toast data.
2645  */
2646 static void
2647 ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
2648                                                   Relation relation, ReorderBufferChange *change)
2649 {
2650         TupleDesc       desc;
2651         int                     natt;
2652         Datum      *attrs;
2653         bool       *isnull;
2654         bool       *free;
2655         HeapTuple       tmphtup;
2656         Relation        toast_rel;
2657         TupleDesc       toast_desc;
2658         MemoryContext oldcontext;
2659         ReorderBufferTupleBuf *newtup;
2660
2661         /* no toast tuples changed */
2662         if (txn->toast_hash == NULL)
2663                 return;
2664
2665         oldcontext = MemoryContextSwitchTo(rb->context);
2666
2667         /* we should only have toast tuples in an INSERT or UPDATE */
2668         Assert(change->data.tp.newtuple);
2669
2670         desc = RelationGetDescr(relation);
2671
2672         toast_rel = RelationIdGetRelation(relation->rd_rel->reltoastrelid);
2673         toast_desc = RelationGetDescr(toast_rel);
2674
2675         /* should we allocate from stack instead? */
2676         attrs = palloc0(sizeof(Datum) * desc->natts);
2677         isnull = palloc0(sizeof(bool) * desc->natts);
2678         free = palloc0(sizeof(bool) * desc->natts);
2679
2680         newtup = change->data.tp.newtuple;
2681
2682         heap_deform_tuple(&newtup->tuple, desc, attrs, isnull);
2683
2684         for (natt = 0; natt < desc->natts; natt++)
2685         {
2686                 Form_pg_attribute attr = desc->attrs[natt];
2687                 ReorderBufferToastEnt *ent;
2688                 struct varlena *varlena;
2689
2690                 /* va_rawsize is the size of the original datum -- including header */
2691                 struct varatt_external toast_pointer;
2692                 struct varatt_indirect redirect_pointer;
2693                 struct varlena *new_datum = NULL;
2694                 struct varlena *reconstructed;
2695                 dlist_iter      it;
2696                 Size            data_done = 0;
2697
2698                 /* system columns aren't toasted */
2699                 if (attr->attnum < 0)
2700                         continue;
2701
2702                 if (attr->attisdropped)
2703                         continue;
2704
2705                 /* not a varlena datatype */
2706                 if (attr->attlen != -1)
2707                         continue;
2708
2709                 /* no data */
2710                 if (isnull[natt])
2711                         continue;
2712
2713                 /* ok, we know we have a toast datum */
2714                 varlena = (struct varlena *) DatumGetPointer(attrs[natt]);
2715
2716                 /* no need to do anything if the tuple isn't external */
2717                 if (!VARATT_IS_EXTERNAL(varlena))
2718                         continue;
2719
2720                 VARATT_EXTERNAL_GET_POINTER(toast_pointer, varlena);
2721
2722                 /*
2723                  * Check whether the toast tuple changed, replace if so.
2724                  */
2725                 ent = (ReorderBufferToastEnt *)
2726                         hash_search(txn->toast_hash,
2727                                                 (void *) &toast_pointer.va_valueid,
2728                                                 HASH_FIND,
2729                                                 NULL);
2730                 if (ent == NULL)
2731                         continue;
2732
2733                 new_datum =
2734                         (struct varlena *) palloc0(INDIRECT_POINTER_SIZE);
2735
2736                 free[natt] = true;
2737
2738                 reconstructed = palloc0(toast_pointer.va_rawsize);
2739
2740                 ent->reconstructed = reconstructed;
2741
2742                 /* stitch toast tuple back together from its parts */
2743                 dlist_foreach(it, &ent->chunks)
2744                 {
2745                         bool            isnull;
2746                         ReorderBufferChange *cchange;
2747                         ReorderBufferTupleBuf *ctup;
2748                         Pointer         chunk;
2749
2750                         cchange = dlist_container(ReorderBufferChange, node, it.cur);
2751                         ctup = cchange->data.tp.newtuple;
2752                         chunk = DatumGetPointer(
2753                                                   fastgetattr(&ctup->tuple, 3, toast_desc, &isnull));
2754
2755                         Assert(!isnull);
2756                         Assert(!VARATT_IS_EXTERNAL(chunk));
2757                         Assert(!VARATT_IS_SHORT(chunk));
2758
2759                         memcpy(VARDATA(reconstructed) + data_done,
2760                                    VARDATA(chunk),
2761                                    VARSIZE(chunk) - VARHDRSZ);
2762                         data_done += VARSIZE(chunk) - VARHDRSZ;
2763                 }
2764                 Assert(data_done == toast_pointer.va_extsize);
2765
2766                 /* make sure its marked as compressed or not */
2767                 if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
2768                         SET_VARSIZE_COMPRESSED(reconstructed, data_done + VARHDRSZ);
2769                 else
2770                         SET_VARSIZE(reconstructed, data_done + VARHDRSZ);
2771
2772                 memset(&redirect_pointer, 0, sizeof(redirect_pointer));
2773                 redirect_pointer.pointer = reconstructed;
2774
2775                 SET_VARTAG_EXTERNAL(new_datum, VARTAG_INDIRECT);
2776                 memcpy(VARDATA_EXTERNAL(new_datum), &redirect_pointer,
2777                            sizeof(redirect_pointer));
2778
2779                 attrs[natt] = PointerGetDatum(new_datum);
2780         }
2781
2782         /*
2783          * Build tuple in separate memory & copy tuple back into the tuplebuf
2784          * passed to the output plugin. We can't directly heap_fill_tuple() into
2785          * the tuplebuf because attrs[] will point back into the current content.
2786          */
2787         tmphtup = heap_form_tuple(desc, attrs, isnull);
2788         Assert(newtup->tuple.t_len <= MaxHeapTupleSize);
2789         Assert(ReorderBufferTupleBufData(newtup) == newtup->tuple.t_data);
2790
2791         memcpy(newtup->tuple.t_data, tmphtup->t_data, tmphtup->t_len);
2792         newtup->tuple.t_len = tmphtup->t_len;
2793
2794         /*
2795          * free resources we won't further need, more persistent stuff will be
2796          * free'd in ReorderBufferToastReset().
2797          */
2798         RelationClose(toast_rel);
2799         pfree(tmphtup);
2800         for (natt = 0; natt < desc->natts; natt++)
2801         {
2802                 if (free[natt])
2803                         pfree(DatumGetPointer(attrs[natt]));
2804         }
2805         pfree(attrs);
2806         pfree(free);
2807         pfree(isnull);
2808
2809         MemoryContextSwitchTo(oldcontext);
2810 }
2811
2812 /*
2813  * Free all resources allocated for toast reconstruction.
2814  */
2815 static void
2816 ReorderBufferToastReset(ReorderBuffer *rb, ReorderBufferTXN *txn)
2817 {
2818         HASH_SEQ_STATUS hstat;
2819         ReorderBufferToastEnt *ent;
2820
2821         if (txn->toast_hash == NULL)
2822                 return;
2823
2824         /* sequentially walk over the hash and free everything */
2825         hash_seq_init(&hstat, txn->toast_hash);
2826         while ((ent = (ReorderBufferToastEnt *) hash_seq_search(&hstat)) != NULL)
2827         {
2828                 dlist_mutable_iter it;
2829
2830                 if (ent->reconstructed != NULL)
2831                         pfree(ent->reconstructed);
2832
2833                 dlist_foreach_modify(it, &ent->chunks)
2834                 {
2835                         ReorderBufferChange *change =
2836                         dlist_container(ReorderBufferChange, node, it.cur);
2837
2838                         dlist_delete(&change->node);
2839                         ReorderBufferReturnChange(rb, change);
2840                 }
2841         }
2842
2843         hash_destroy(txn->toast_hash);
2844         txn->toast_hash = NULL;
2845 }
2846
2847
2848 /* ---------------------------------------
2849  * Visibility support for logical decoding
2850  *
2851  *
2852  * Lookup actual cmin/cmax values when using decoding snapshot. We can't
2853  * always rely on stored cmin/cmax values because of two scenarios:
2854  *
2855  * * A tuple got changed multiple times during a single transaction and thus
2856  *       has got a combocid. Combocid's are only valid for the duration of a
2857  *       single transaction.
2858  * * A tuple with a cmin but no cmax (and thus no combocid) got
2859  *       deleted/updated in another transaction than the one which created it
2860  *       which we are looking at right now. As only one of cmin, cmax or combocid
2861  *       is actually stored in the heap we don't have access to the value we
2862  *       need anymore.
2863  *
2864  * To resolve those problems we have a per-transaction hash of (cmin,
2865  * cmax) tuples keyed by (relfilenode, ctid) which contains the actual
2866  * (cmin, cmax) values. That also takes care of combocids by simply
2867  * not caring about them at all. As we have the real cmin/cmax values
2868  * combocids aren't interesting.
2869  *
2870  * As we only care about catalog tuples here the overhead of this
2871  * hashtable should be acceptable.
2872  *
2873  * Heap rewrites complicate this a bit, check rewriteheap.c for
2874  * details.
2875  * -------------------------------------------------------------------------
2876  */
2877
2878 /* struct for qsort()ing mapping files by lsn somewhat efficiently */
2879 typedef struct RewriteMappingFile
2880 {
2881         XLogRecPtr      lsn;
2882         char            fname[MAXPGPATH];
2883 } RewriteMappingFile;
2884
2885 #if NOT_USED
2886 static void
2887 DisplayMapping(HTAB *tuplecid_data)
2888 {
2889         HASH_SEQ_STATUS hstat;
2890         ReorderBufferTupleCidEnt *ent;
2891
2892         hash_seq_init(&hstat, tuplecid_data);
2893         while ((ent = (ReorderBufferTupleCidEnt *) hash_seq_search(&hstat)) != NULL)
2894         {
2895                 elog(DEBUG3, "mapping: node: %u/%u/%u tid: %u/%u cmin: %u, cmax: %u",
2896                          ent->key.relnode.dbNode,
2897                          ent->key.relnode.spcNode,
2898                          ent->key.relnode.relNode,
2899                          BlockIdGetBlockNumber(&ent->key.tid.ip_blkid),
2900                          ent->key.tid.ip_posid,
2901                          ent->cmin,
2902                          ent->cmax
2903                         );
2904         }
2905 }
2906 #endif
2907
2908 /*
2909  * Apply a single mapping file to tuplecid_data.
2910  *
2911  * The mapping file has to have been verified to be a) committed b) for our
2912  * transaction c) applied in LSN order.
2913  */
2914 static void
2915 ApplyLogicalMappingFile(HTAB *tuplecid_data, Oid relid, const char *fname)
2916 {
2917         char            path[MAXPGPATH];
2918         int                     fd;
2919         int                     readBytes;
2920         LogicalRewriteMappingData map;
2921
2922         sprintf(path, "pg_logical/mappings/%s", fname);
2923         fd = OpenTransientFile(path, O_RDONLY | PG_BINARY, 0);
2924         if (fd < 0)
2925                 ereport(ERROR,
2926                                 (errmsg("could not open file \"%s\": %m", path)));
2927
2928         while (true)
2929         {
2930                 ReorderBufferTupleCidKey key;
2931                 ReorderBufferTupleCidEnt *ent;
2932                 ReorderBufferTupleCidEnt *new_ent;
2933                 bool            found;
2934
2935                 /* be careful about padding */
2936                 memset(&key, 0, sizeof(ReorderBufferTupleCidKey));
2937
2938                 /* read all mappings till the end of the file */
2939                 readBytes = read(fd, &map, sizeof(LogicalRewriteMappingData));
2940
2941                 if (readBytes < 0)
2942                         ereport(ERROR,
2943                                         (errcode_for_file_access(),
2944                                          errmsg("could not read file \"%s\": %m",
2945                                                         path)));
2946                 else if (readBytes == 0)        /* EOF */
2947                         break;
2948                 else if (readBytes != sizeof(LogicalRewriteMappingData))
2949                         ereport(ERROR,
2950                                         (errcode_for_file_access(),
2951                                          errmsg("could not read from file \"%s\": read %d instead of %d bytes",
2952                                                         path, readBytes,
2953                                                         (int32) sizeof(LogicalRewriteMappingData))));
2954
2955                 key.relnode = map.old_node;
2956                 ItemPointerCopy(&map.old_tid,
2957                                                 &key.tid);
2958
2959
2960                 ent = (ReorderBufferTupleCidEnt *)
2961                         hash_search(tuplecid_data,
2962                                                 (void *) &key,
2963                                                 HASH_FIND,
2964                                                 NULL);
2965
2966                 /* no existing mapping, no need to update */
2967                 if (!ent)
2968                         continue;
2969
2970                 key.relnode = map.new_node;
2971                 ItemPointerCopy(&map.new_tid,
2972                                                 &key.tid);
2973
2974                 new_ent = (ReorderBufferTupleCidEnt *)
2975                         hash_search(tuplecid_data,
2976                                                 (void *) &key,
2977                                                 HASH_ENTER,
2978                                                 &found);
2979
2980                 if (found)
2981                 {
2982                         /*
2983                          * Make sure the existing mapping makes sense. We sometime update
2984                          * old records that did not yet have a cmax (e.g. pg_class' own
2985                          * entry while rewriting it) during rewrites, so allow that.
2986                          */
2987                         Assert(ent->cmin == InvalidCommandId || ent->cmin == new_ent->cmin);
2988                         Assert(ent->cmax == InvalidCommandId || ent->cmax == new_ent->cmax);
2989                 }
2990                 else
2991                 {
2992                         /* update mapping */
2993                         new_ent->cmin = ent->cmin;
2994                         new_ent->cmax = ent->cmax;
2995                         new_ent->combocid = ent->combocid;
2996                 }
2997         }
2998 }
2999
3000
3001 /*
3002  * Check whether the TransactionOId 'xid' is in the pre-sorted array 'xip'.
3003  */
3004 static bool
3005 TransactionIdInArray(TransactionId xid, TransactionId *xip, Size num)
3006 {
3007         return bsearch(&xid, xip, num,
3008                                    sizeof(TransactionId), xidComparator) != NULL;
3009 }
3010
3011 /*
3012  * qsort() comparator for sorting RewriteMappingFiles in LSN order.
3013  */
3014 static int
3015 file_sort_by_lsn(const void *a_p, const void *b_p)
3016 {
3017         RewriteMappingFile *a = *(RewriteMappingFile **) a_p;
3018         RewriteMappingFile *b = *(RewriteMappingFile **) b_p;
3019
3020         if (a->lsn < b->lsn)
3021                 return -1;
3022         else if (a->lsn > b->lsn)
3023                 return 1;
3024         return 0;
3025 }
3026
3027 /*
3028  * Apply any existing logical remapping files if there are any targeted at our
3029  * transaction for relid.
3030  */
3031 static void
3032 UpdateLogicalMappings(HTAB *tuplecid_data, Oid relid, Snapshot snapshot)
3033 {
3034         DIR                *mapping_dir;
3035         struct dirent *mapping_de;
3036         List       *files = NIL;
3037         ListCell   *file;
3038         RewriteMappingFile **files_a;
3039         size_t          off;
3040         Oid                     dboid = IsSharedRelation(relid) ? InvalidOid : MyDatabaseId;
3041
3042         mapping_dir = AllocateDir("pg_logical/mappings");
3043         while ((mapping_de = ReadDir(mapping_dir, "pg_logical/mappings")) != NULL)
3044         {
3045                 Oid                     f_dboid;
3046                 Oid                     f_relid;
3047                 TransactionId f_mapped_xid;
3048                 TransactionId f_create_xid;
3049                 XLogRecPtr      f_lsn;
3050                 uint32          f_hi,
3051                                         f_lo;
3052                 RewriteMappingFile *f;
3053
3054                 if (strcmp(mapping_de->d_name, ".") == 0 ||
3055                         strcmp(mapping_de->d_name, "..") == 0)
3056                         continue;
3057
3058                 /* Ignore files that aren't ours */
3059                 if (strncmp(mapping_de->d_name, "map-", 4) != 0)
3060                         continue;
3061
3062                 if (sscanf(mapping_de->d_name, LOGICAL_REWRITE_FORMAT,
3063                                    &f_dboid, &f_relid, &f_hi, &f_lo,
3064                                    &f_mapped_xid, &f_create_xid) != 6)
3065                         elog(ERROR, "could not parse filename \"%s\"", mapping_de->d_name);
3066
3067                 f_lsn = ((uint64) f_hi) << 32 | f_lo;
3068
3069                 /* mapping for another database */
3070                 if (f_dboid != dboid)
3071                         continue;
3072
3073                 /* mapping for another relation */
3074                 if (f_relid != relid)
3075                         continue;
3076
3077                 /* did the creating transaction abort? */
3078                 if (!TransactionIdDidCommit(f_create_xid))
3079                         continue;
3080
3081                 /* not for our transaction */
3082                 if (!TransactionIdInArray(f_mapped_xid, snapshot->subxip, snapshot->subxcnt))
3083                         continue;
3084
3085                 /* ok, relevant, queue for apply */
3086                 f = palloc(sizeof(RewriteMappingFile));
3087                 f->lsn = f_lsn;
3088                 strcpy(f->fname, mapping_de->d_name);
3089                 files = lappend(files, f);
3090         }
3091         FreeDir(mapping_dir);
3092
3093         /* build array we can easily sort */
3094         files_a = palloc(list_length(files) * sizeof(RewriteMappingFile *));
3095         off = 0;
3096         foreach(file, files)
3097         {
3098                 files_a[off++] = lfirst(file);
3099         }
3100
3101         /* sort files so we apply them in LSN order */
3102         qsort(files_a, list_length(files), sizeof(RewriteMappingFile *),
3103                   file_sort_by_lsn);
3104
3105         for (off = 0; off < list_length(files); off++)
3106         {
3107                 RewriteMappingFile *f = files_a[off];
3108
3109                 elog(DEBUG1, "applying mapping: \"%s\" in %u", f->fname,
3110                          snapshot->subxip[0]);
3111                 ApplyLogicalMappingFile(tuplecid_data, relid, f->fname);
3112                 pfree(f);
3113         }
3114 }
3115
3116 /*
3117  * Lookup cmin/cmax of a tuple, during logical decoding where we can't rely on
3118  * combocids.
3119  */
3120 bool
3121 ResolveCminCmaxDuringDecoding(HTAB *tuplecid_data,
3122                                                           Snapshot snapshot,
3123                                                           HeapTuple htup, Buffer buffer,
3124                                                           CommandId *cmin, CommandId *cmax)
3125 {
3126         ReorderBufferTupleCidKey key;
3127         ReorderBufferTupleCidEnt *ent;
3128         ForkNumber      forkno;
3129         BlockNumber blockno;
3130         bool            updated_mapping = false;
3131
3132         /* be careful about padding */
3133         memset(&key, 0, sizeof(key));
3134
3135         Assert(!BufferIsLocal(buffer));
3136
3137         /*
3138          * get relfilenode from the buffer, no convenient way to access it other
3139          * than that.
3140          */
3141         BufferGetTag(buffer, &key.relnode, &forkno, &blockno);
3142
3143         /* tuples can only be in the main fork */
3144         Assert(forkno == MAIN_FORKNUM);
3145         Assert(blockno == ItemPointerGetBlockNumber(&htup->t_self));
3146
3147         ItemPointerCopy(&htup->t_self,
3148                                         &key.tid);
3149
3150 restart:
3151         ent = (ReorderBufferTupleCidEnt *)
3152                 hash_search(tuplecid_data,
3153                                         (void *) &key,
3154                                         HASH_FIND,
3155                                         NULL);
3156
3157         /*
3158          * failed to find a mapping, check whether the table was rewritten and
3159          * apply mapping if so, but only do that once - there can be no new
3160          * mappings while we are in here since we have to hold a lock on the
3161          * relation.
3162          */
3163         if (ent == NULL && !updated_mapping)
3164         {
3165                 UpdateLogicalMappings(tuplecid_data, htup->t_tableOid, snapshot);
3166                 /* now check but don't update for a mapping again */
3167                 updated_mapping = true;
3168                 goto restart;
3169         }
3170         else if (ent == NULL)
3171                 return false;
3172
3173         if (cmin)
3174                 *cmin = ent->cmin;
3175         if (cmax)
3176                 *cmax = ent->cmax;
3177         return true;
3178 }