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