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