reorderbuffer.c File Reference

#include "postgres.h"
#include <unistd.h>
#include <sys/stat.h>
#include "access/detoast.h"
#include "access/heapam.h"
#include "access/rewriteheap.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "catalog/catalog.h"
#include "common/int.h"
#include "lib/binaryheap.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/logical.h"
#include "replication/reorderbuffer.h"
#include "replication/slot.h"
#include "replication/snapbuild.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
#include "storage/sinval.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/relfilenumbermap.h"

Include dependency graph for reorderbuffer.c:

Go to the source code of this file.

Data Structures
struct	ReorderBufferTXNByIdEnt
struct	ReorderBufferTupleCidKey
struct	ReorderBufferTupleCidEnt
struct	TXNEntryFile
struct	ReorderBufferIterTXNEntry
struct	ReorderBufferIterTXNState
struct	ReorderBufferToastEnt
struct	ReorderBufferDiskChange
struct	RewriteMappingFile

Macros
#define	IsSpecInsert(action)
#define	IsSpecConfirmOrAbort(action)
#define	IsInsertOrUpdate(action)
#define	CHANGES_THRESHOLD   100

Typedefs
typedef struct ReorderBufferTXNByIdEnt	ReorderBufferTXNByIdEnt
typedef struct ReorderBufferTupleCidKey	ReorderBufferTupleCidKey
typedef struct ReorderBufferTupleCidEnt	ReorderBufferTupleCidEnt
typedef struct TXNEntryFile	TXNEntryFile
typedef struct ReorderBufferIterTXNEntry	ReorderBufferIterTXNEntry
typedef struct ReorderBufferIterTXNState	ReorderBufferIterTXNState
typedef struct ReorderBufferToastEnt	ReorderBufferToastEnt
typedef struct ReorderBufferDiskChange	ReorderBufferDiskChange
typedef struct RewriteMappingFile	RewriteMappingFile

Functions
static ReorderBufferTXN *	ReorderBufferGetTXN (ReorderBuffer *rb)
static void	ReorderBufferReturnTXN (ReorderBuffer *rb, ReorderBufferTXN *txn)
static ReorderBufferTXN *	ReorderBufferTXNByXid (ReorderBuffer *rb, TransactionId xid, bool create, bool *is_new, XLogRecPtr lsn, bool create_as_top)
static void	ReorderBufferTransferSnapToParent (ReorderBufferTXN *txn, ReorderBufferTXN *subtxn)
static void	AssertTXNLsnOrder (ReorderBuffer *rb)
static void	ReorderBufferIterTXNInit (ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferIterTXNState *volatile *iter_state)
static ReorderBufferChange *	ReorderBufferIterTXNNext (ReorderBuffer *rb, ReorderBufferIterTXNState *state)
static void	ReorderBufferIterTXNFinish (ReorderBuffer *rb, ReorderBufferIterTXNState *state)
static void	ReorderBufferExecuteInvalidations (uint32 nmsgs, SharedInvalidationMessage *msgs)
static void	ReorderBufferCheckMemoryLimit (ReorderBuffer *rb)
static void	ReorderBufferSerializeTXN (ReorderBuffer *rb, ReorderBufferTXN *txn)
static void	ReorderBufferSerializeChange (ReorderBuffer *rb, ReorderBufferTXN *txn, int fd, ReorderBufferChange *change)
static Size	ReorderBufferRestoreChanges (ReorderBuffer *rb, ReorderBufferTXN *txn, TXNEntryFile *file, XLogSegNo *segno)
static void	ReorderBufferRestoreChange (ReorderBuffer *rb, ReorderBufferTXN *txn, char *data)
static void	ReorderBufferRestoreCleanup (ReorderBuffer *rb, ReorderBufferTXN *txn)
static void	ReorderBufferTruncateTXN (ReorderBuffer *rb, ReorderBufferTXN *txn, bool txn_prepared)
static void	ReorderBufferCleanupSerializedTXNs (const char *slotname)
static void	ReorderBufferSerializedPath (char *path, ReplicationSlot *slot, TransactionId xid, XLogSegNo segno)
static int	ReorderBufferTXNSizeCompare (const pairingheap_node *a, const pairingheap_node *b, void *arg)
static void	ReorderBufferFreeSnap (ReorderBuffer *rb, Snapshot snap)
static Snapshot	ReorderBufferCopySnap (ReorderBuffer *rb, Snapshot orig_snap, ReorderBufferTXN *txn, CommandId cid)
static bool	ReorderBufferCanStream (ReorderBuffer *rb)
static bool	ReorderBufferCanStartStreaming (ReorderBuffer *rb)
static void	ReorderBufferStreamTXN (ReorderBuffer *rb, ReorderBufferTXN *txn)
static void	ReorderBufferStreamCommit (ReorderBuffer *rb, ReorderBufferTXN *txn)
static void	ReorderBufferToastInitHash (ReorderBuffer *rb, ReorderBufferTXN *txn)
static void	ReorderBufferToastReset (ReorderBuffer *rb, ReorderBufferTXN *txn)
static void	ReorderBufferToastReplace (ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
static void	ReorderBufferToastAppendChunk (ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
static Size	ReorderBufferChangeSize (ReorderBufferChange *change)
static void	ReorderBufferChangeMemoryUpdate (ReorderBuffer *rb, ReorderBufferChange *change, ReorderBufferTXN *txn, bool addition, Size sz)
ReorderBuffer *	ReorderBufferAllocate (void)
void	ReorderBufferFree (ReorderBuffer *rb)
ReorderBufferChange *	ReorderBufferGetChange (ReorderBuffer *rb)
void	ReorderBufferReturnChange (ReorderBuffer *rb, ReorderBufferChange *change, bool upd_mem)
HeapTuple	ReorderBufferGetTupleBuf (ReorderBuffer *rb, Size tuple_len)
void	ReorderBufferReturnTupleBuf (HeapTuple tuple)
Oid *	ReorderBufferGetRelids (ReorderBuffer *rb, int nrelids)
void	ReorderBufferReturnRelids (ReorderBuffer *rb, Oid *relids)
static void	ReorderBufferProcessPartialChange (ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferChange *change, bool toast_insert)
void	ReorderBufferQueueChange (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
void	ReorderBufferQueueMessage (ReorderBuffer *rb, TransactionId xid, Snapshot snap, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
static void	AssertChangeLsnOrder (ReorderBufferTXN *txn)
ReorderBufferTXN *	ReorderBufferGetOldestTXN (ReorderBuffer *rb)
TransactionId	ReorderBufferGetOldestXmin (ReorderBuffer *rb)
void	ReorderBufferSetRestartPoint (ReorderBuffer *rb, XLogRecPtr ptr)
void	ReorderBufferAssignChild (ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
void	ReorderBufferCommitChild (ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
static int	ReorderBufferIterCompare (Datum a, Datum b, void *arg)
static void	ReorderBufferCleanupTXN (ReorderBuffer *rb, ReorderBufferTXN *txn)
static void	ReorderBufferBuildTupleCidHash (ReorderBuffer *rb, ReorderBufferTXN *txn)
static void	SetupCheckXidLive (TransactionId xid)
static void	ReorderBufferApplyChange (ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change, bool streaming)
static void	ReorderBufferApplyTruncate (ReorderBuffer *rb, ReorderBufferTXN *txn, int nrelations, Relation *relations, ReorderBufferChange *change, bool streaming)
static void	ReorderBufferApplyMessage (ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferChange *change, bool streaming)
static void	ReorderBufferSaveTXNSnapshot (ReorderBuffer *rb, ReorderBufferTXN *txn, Snapshot snapshot_now, CommandId command_id)
static void	ReorderBufferResetTXN (ReorderBuffer *rb, ReorderBufferTXN *txn, Snapshot snapshot_now, CommandId command_id, XLogRecPtr last_lsn, ReorderBufferChange *specinsert)
static void	ReorderBufferProcessTXN (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr commit_lsn, volatile Snapshot snapshot_now, volatile CommandId command_id, bool streaming)
static void	ReorderBufferReplay (ReorderBufferTXN *txn, ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
void	ReorderBufferCommit (ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
bool	ReorderBufferRememberPrepareInfo (ReorderBuffer *rb, TransactionId xid, XLogRecPtr prepare_lsn, XLogRecPtr end_lsn, TimestampTz prepare_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
void	ReorderBufferSkipPrepare (ReorderBuffer *rb, TransactionId xid)
void	ReorderBufferPrepare (ReorderBuffer *rb, TransactionId xid, char *gid)
void	ReorderBufferFinishPrepared (ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, XLogRecPtr two_phase_at, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn, char *gid, bool is_commit)
void	ReorderBufferAbort (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, TimestampTz abort_time)
void	ReorderBufferAbortOld (ReorderBuffer *rb, TransactionId oldestRunningXid)
void	ReorderBufferForget (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void	ReorderBufferInvalidate (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void	ReorderBufferImmediateInvalidation (ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
void	ReorderBufferProcessXid (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void	ReorderBufferAddSnapshot (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
void	ReorderBufferSetBaseSnapshot (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
void	ReorderBufferAddNewCommandId (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)
void	ReorderBufferAddNewTupleCids (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
void	ReorderBufferAddInvalidations (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
void	ReorderBufferXidSetCatalogChanges (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
TransactionId *	ReorderBufferGetCatalogChangesXacts (ReorderBuffer *rb)
bool	ReorderBufferXidHasCatalogChanges (ReorderBuffer *rb, TransactionId xid)
bool	ReorderBufferXidHasBaseSnapshot (ReorderBuffer *rb, TransactionId xid)
static void	ReorderBufferSerializeReserve (ReorderBuffer *rb, Size sz)
static ReorderBufferTXN *	ReorderBufferLargestTXN (ReorderBuffer *rb)
static ReorderBufferTXN *	ReorderBufferLargestStreamableTopTXN (ReorderBuffer *rb)
void	StartupReorderBuffer (void)
static void	ApplyLogicalMappingFile (HTAB *tuplecid_data, Oid relid, const char *fname)
static bool	TransactionIdInArray (TransactionId xid, TransactionId *xip, Size num)
static int	file_sort_by_lsn (const ListCell *a_p, const ListCell *b_p)
static void	UpdateLogicalMappings (HTAB *tuplecid_data, Oid relid, Snapshot snapshot)
bool	ResolveCminCmaxDuringDecoding (HTAB *tuplecid_data, Snapshot snapshot, HeapTuple htup, Buffer buffer, CommandId *cmin, CommandId *cmax)

Variables
int	logical_decoding_work_mem
static const Size	max_changes_in_memory = 4096
int	debug_logical_replication_streaming = DEBUG_LOGICAL_REP_STREAMING_BUFFERED

Macro Definition Documentation

CHANGES_THRESHOLD

#define CHANGES_THRESHOLD   100

IsInsertOrUpdate

#define IsInsertOrUpdate

(

action

)

Value:

( \
    (((action) == REORDER_BUFFER_CHANGE_INSERT) || \
    ((action) == REORDER_BUFFER_CHANGE_UPDATE) || \
    ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT)) \
)

Definition at line 193 of file reorderbuffer.c.

IsSpecConfirmOrAbort

#define IsSpecConfirmOrAbort

(

action

)

Value:

( \
    (((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM) || \
    ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT)) \
)

Definition at line 188 of file reorderbuffer.c.

IsSpecInsert

#define IsSpecInsert

(

action

)

Value:

( \
    ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT) \
)

Definition at line 184 of file reorderbuffer.c.

Typedef Documentation

ReorderBufferDiskChange

typedef struct ReorderBufferDiskChange ReorderBufferDiskChange

ReorderBufferIterTXNEntry

typedef struct ReorderBufferIterTXNEntry ReorderBufferIterTXNEntry

ReorderBufferIterTXNState

typedef struct ReorderBufferIterTXNState ReorderBufferIterTXNState

ReorderBufferToastEnt

typedef struct ReorderBufferToastEnt ReorderBufferToastEnt

ReorderBufferTupleCidEnt

typedef struct ReorderBufferTupleCidEnt ReorderBufferTupleCidEnt

ReorderBufferTupleCidKey

typedef struct ReorderBufferTupleCidKey ReorderBufferTupleCidKey

ReorderBufferTXNByIdEnt

typedef struct ReorderBufferTXNByIdEnt ReorderBufferTXNByIdEnt

RewriteMappingFile

typedef struct RewriteMappingFile RewriteMappingFile

TXNEntryFile

typedef struct TXNEntryFile TXNEntryFile

Function Documentation

ApplyLogicalMappingFile()

static void ApplyLogicalMappingFile ( HTAB *  tuplecid_data, Oid  relid, const char *  fname  )

static

Definition at line 5077 of file reorderbuffer.c.

{
    char        path[MAXPGPATH];
    int         fd;
    int         readBytes;
    LogicalRewriteMappingData map;
 
    sprintf(path, "%s/%s", PG_LOGICAL_MAPPINGS_DIR, fname);
    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
    if (fd < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", path)));
 
    while (true)
    {
        ReorderBufferTupleCidKey key;
        ReorderBufferTupleCidEnt *ent;
        ReorderBufferTupleCidEnt *new_ent;
        bool        found;
 
        /* be careful about padding */
        memset(&key, 0, sizeof(ReorderBufferTupleCidKey));
 
        /* read all mappings till the end of the file */
        pgstat_report_wait_start(WAIT_EVENT_REORDER_LOGICAL_MAPPING_READ);
        readBytes = read(fd, &map, sizeof(LogicalRewriteMappingData));
        pgstat_report_wait_end();
 
        if (readBytes < 0)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m",
                            path)));
        else if (readBytes == 0)    /* EOF */
            break;
        else if (readBytes != sizeof(LogicalRewriteMappingData))
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read from file \"%s\": read %d instead of %d bytes",
                            path, readBytes,
                            (int32) sizeof(LogicalRewriteMappingData))));
 
        key.rlocator = map.old_locator;
        ItemPointerCopy(&map.old_tid,
                        &key.tid);
 
 
        ent = (ReorderBufferTupleCidEnt *)
            hash_search(tuplecid_data, &key, HASH_FIND, NULL);
 
        /* no existing mapping, no need to update */
        if (!ent)
            continue;
 
        key.rlocator = map.new_locator;
        ItemPointerCopy(&map.new_tid,
                        &key.tid);
 
        new_ent = (ReorderBufferTupleCidEnt *)
            hash_search(tuplecid_data, &key, HASH_ENTER, &found);
 
        if (found)
        {
            /*
             * Make sure the existing mapping makes sense. We sometime update
             * old records that did not yet have a cmax (e.g. pg_class' own
             * entry while rewriting it) during rewrites, so allow that.
             */
            Assert(ent->cmin == InvalidCommandId || ent->cmin == new_ent->cmin);
            Assert(ent->cmax == InvalidCommandId || ent->cmax == new_ent->cmax);
        }
        else
        {
            /* update mapping */
            new_ent->cmin = ent->cmin;
            new_ent->cmax = ent->cmax;
            new_ent->combocid = ent->combocid;
        }
    }
 
    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", path)));
}

References Assert, CloseTransientFile(), ReorderBufferTupleCidEnt::cmax, ReorderBufferTupleCidEnt::cmin, ReorderBufferTupleCidEnt::combocid, ereport, errcode_for_file_access(), errmsg(), ERROR, fd(), HASH_ENTER, HASH_FIND, hash_search(), InvalidCommandId, ItemPointerCopy(), sort-test::key, MAXPGPATH, LogicalRewriteMappingData::new_locator, LogicalRewriteMappingData::new_tid, LogicalRewriteMappingData::old_locator, LogicalRewriteMappingData::old_tid, OpenTransientFile(), PG_BINARY, PG_LOGICAL_MAPPINGS_DIR, pgstat_report_wait_end(), pgstat_report_wait_start(), read, sprintf, and tuplecid_data.

Referenced by UpdateLogicalMappings().

AssertChangeLsnOrder()

static void AssertChangeLsnOrder ( ReorderBufferTXN *  txn )

static

Definition at line 986 of file reorderbuffer.c.

{
#ifdef USE_ASSERT_CHECKING
    dlist_iter  iter;
    XLogRecPtr  prev_lsn = txn->first_lsn;
 
    dlist_foreach(iter, &txn->changes)
    {
        ReorderBufferChange *cur_change;
 
        cur_change = dlist_container(ReorderBufferChange, node, iter.cur);
 
        Assert(txn->first_lsn != InvalidXLogRecPtr);
        Assert(cur_change->lsn != InvalidXLogRecPtr);
        Assert(txn->first_lsn <= cur_change->lsn);
 
        if (txn->end_lsn != InvalidXLogRecPtr)
            Assert(cur_change->lsn <= txn->end_lsn);
 
        Assert(prev_lsn <= cur_change->lsn);
 
        prev_lsn = cur_change->lsn;
    }
#endif
}

References Assert, ReorderBufferTXN::changes, dlist_iter::cur, dlist_container, dlist_foreach, ReorderBufferTXN::end_lsn, ReorderBufferTXN::first_lsn, InvalidXLogRecPtr, and ReorderBufferChange::lsn.

Referenced by ReorderBufferIterTXNInit().

AssertTXNLsnOrder()

static void AssertTXNLsnOrder ( ReorderBuffer *  rb )

static

Definition at line 915 of file reorderbuffer.c.

{
#ifdef USE_ASSERT_CHECKING
    LogicalDecodingContext *ctx = rb->private_data;
    dlist_iter  iter;
    XLogRecPtr  prev_first_lsn = InvalidXLogRecPtr;
    XLogRecPtr  prev_base_snap_lsn = InvalidXLogRecPtr;
 
    /*
     * Skip the verification if we don't reach the LSN at which we start
     * decoding the contents of transactions yet because until we reach the
     * LSN, we could have transactions that don't have the association between
     * the top-level transaction and subtransaction yet and consequently have
     * the same LSN.  We don't guarantee this association until we try to
     * decode the actual contents of transaction. The ordering of the records
     * prior to the start_decoding_at LSN should have been checked before the
     * restart.
     */
    if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, ctx->reader->EndRecPtr))
        return;
 
    dlist_foreach(iter, &rb->toplevel_by_lsn)
    {
        ReorderBufferTXN *cur_txn = dlist_container(ReorderBufferTXN, node,
                                                    iter.cur);
 
        /* start LSN must be set */
        Assert(cur_txn->first_lsn != InvalidXLogRecPtr);
 
        /* If there is an end LSN, it must be higher than start LSN */
        if (cur_txn->end_lsn != InvalidXLogRecPtr)
            Assert(cur_txn->first_lsn <= cur_txn->end_lsn);
 
        /* Current initial LSN must be strictly higher than previous */
        if (prev_first_lsn != InvalidXLogRecPtr)
            Assert(prev_first_lsn < cur_txn->first_lsn);
 
        /* known-as-subtxn txns must not be listed */
        Assert(!rbtxn_is_known_subxact(cur_txn));
 
        prev_first_lsn = cur_txn->first_lsn;
    }
 
    dlist_foreach(iter, &rb->txns_by_base_snapshot_lsn)
    {
        ReorderBufferTXN *cur_txn = dlist_container(ReorderBufferTXN,
                                                    base_snapshot_node,
                                                    iter.cur);
 
        /* base snapshot (and its LSN) must be set */
        Assert(cur_txn->base_snapshot != NULL);
        Assert(cur_txn->base_snapshot_lsn != InvalidXLogRecPtr);
 
        /* current LSN must be strictly higher than previous */
        if (prev_base_snap_lsn != InvalidXLogRecPtr)
            Assert(prev_base_snap_lsn < cur_txn->base_snapshot_lsn);
 
        /* known-as-subtxn txns must not be listed */
        Assert(!rbtxn_is_known_subxact(cur_txn));
 
        prev_base_snap_lsn = cur_txn->base_snapshot_lsn;
    }
#endif
}

References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::base_snapshot_lsn, dlist_iter::cur, dlist_container, dlist_foreach, ReorderBufferTXN::end_lsn, XLogReaderState::EndRecPtr, ReorderBufferTXN::first_lsn, InvalidXLogRecPtr, ReorderBuffer::private_data, rbtxn_is_known_subxact, LogicalDecodingContext::reader, SnapBuildXactNeedsSkip(), LogicalDecodingContext::snapshot_builder, ReorderBuffer::toplevel_by_lsn, and ReorderBuffer::txns_by_base_snapshot_lsn.

Referenced by ReorderBufferAssignChild(), ReorderBufferGetOldestTXN(), ReorderBufferGetOldestXmin(), ReorderBufferSetBaseSnapshot(), and ReorderBufferTXNByXid().

file_sort_by_lsn()

static int file_sort_by_lsn ( const ListCell *  a_p, const ListCell *  b_p  )

static

Definition at line 5179 of file reorderbuffer.c.

{
    RewriteMappingFile *a = (RewriteMappingFile *) lfirst(a_p);
    RewriteMappingFile *b = (RewriteMappingFile *) lfirst(b_p);
 
    return pg_cmp_u64(a->lsn, b->lsn);
}

References a, b, lfirst, and pg_cmp_u64().

Referenced by UpdateLogicalMappings().

ReorderBufferAbort()

void ReorderBufferAbort	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn,
		TimestampTz	abort_time
	)

Definition at line 2926 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
 
    /* unknown, nothing to remove */
    if (txn == NULL)
        return;
 
    txn->xact_time.abort_time = abort_time;
 
    /* For streamed transactions notify the remote node about the abort. */
    if (rbtxn_is_streamed(txn))
    {
        rb->stream_abort(rb, txn, lsn);
 
        /*
         * We might have decoded changes for this transaction that could load
         * the cache as per the current transaction's view (consider DDL's
         * happened in this transaction). We don't want the decoding of future
         * transactions to use those cache entries so execute invalidations.
         */
        if (txn->ninvalidations > 0)
            ReorderBufferImmediateInvalidation(rb, txn->ninvalidations,
                                               txn->invalidations);
    }
 
    /* cosmetic... */
    txn->final_lsn = lsn;
 
    /* remove potential on-disk data, and deallocate */
    ReorderBufferCleanupTXN(rb, txn);
}

References ReorderBufferTXN::abort_time, ReorderBufferTXN::final_lsn, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBufferImmediateInvalidation(), ReorderBufferTXNByXid(), ReorderBuffer::stream_abort, and ReorderBufferTXN::xact_time.

Referenced by DecodeAbort().

ReorderBufferAbortOld()

void ReorderBufferAbortOld	(	ReorderBuffer *	rb,
		TransactionId	oldestRunningXid
	)

Definition at line 2971 of file reorderbuffer.c.

{
    dlist_mutable_iter it;
 
    /*
     * Iterate through all (potential) toplevel TXNs and abort all that are
     * older than what possibly can be running. Once we've found the first
     * that is alive we stop, there might be some that acquired an xid earlier
     * but started writing later, but it's unlikely and they will be cleaned
     * up in a later call to this function.
     */
    dlist_foreach_modify(it, &rb->toplevel_by_lsn)
    {
        ReorderBufferTXN *txn;
 
        txn = dlist_container(ReorderBufferTXN, node, it.cur);
 
        if (TransactionIdPrecedes(txn->xid, oldestRunningXid))
        {
            elog(DEBUG2, "aborting old transaction %u", txn->xid);
 
            /* Notify the remote node about the crash/immediate restart. */
            if (rbtxn_is_streamed(txn))
                rb->stream_abort(rb, txn, InvalidXLogRecPtr);
 
            /* remove potential on-disk data, and deallocate this tx */
            ReorderBufferCleanupTXN(rb, txn);
        }
        else
            return;
    }
}

References dlist_mutable_iter::cur, DEBUG2, dlist_container, dlist_foreach_modify, elog, InvalidXLogRecPtr, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBuffer::stream_abort, ReorderBuffer::toplevel_by_lsn, TransactionIdPrecedes(), and ReorderBufferTXN::xid.

Referenced by standby_decode().

ReorderBufferAddInvalidations()

void ReorderBufferAddInvalidations	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn,
		Size	nmsgs,
		SharedInvalidationMessage *	msgs
	)

Definition at line 3324 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
    MemoryContext oldcontext;
    ReorderBufferChange *change;
 
    txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
    oldcontext = MemoryContextSwitchTo(rb->context);
 
    /*
     * Collect all the invalidations under the top transaction, if available,
     * so that we can execute them all together.  See comments atop this
     * function.
     */
    txn = rbtxn_get_toptxn(txn);
 
    Assert(nmsgs > 0);
 
    /* Accumulate invalidations. */
    if (txn->ninvalidations == 0)
    {
        txn->ninvalidations = nmsgs;
        txn->invalidations = (SharedInvalidationMessage *)
            palloc(sizeof(SharedInvalidationMessage) * nmsgs);
        memcpy(txn->invalidations, msgs,
               sizeof(SharedInvalidationMessage) * nmsgs);
    }
    else
    {
        txn->invalidations = (SharedInvalidationMessage *)
            repalloc(txn->invalidations, sizeof(SharedInvalidationMessage) *
                     (txn->ninvalidations + nmsgs));
 
        memcpy(txn->invalidations + txn->ninvalidations, msgs,
               nmsgs * sizeof(SharedInvalidationMessage));
        txn->ninvalidations += nmsgs;
    }
 
    change = ReorderBufferGetChange(rb);
    change->action = REORDER_BUFFER_CHANGE_INVALIDATION;
    change->data.inval.ninvalidations = nmsgs;
    change->data.inval.invalidations = (SharedInvalidationMessage *)
        palloc(sizeof(SharedInvalidationMessage) * nmsgs);
    memcpy(change->data.inval.invalidations, msgs,
           sizeof(SharedInvalidationMessage) * nmsgs);
 
    ReorderBufferQueueChange(rb, xid, lsn, change, false);
 
    MemoryContextSwitchTo(oldcontext);
}

References ReorderBufferChange::action, Assert, ReorderBuffer::context, ReorderBufferChange::data, ReorderBufferChange::inval, ReorderBufferTXN::invalidations, MemoryContextSwitchTo(), ReorderBufferTXN::ninvalidations, palloc(), rbtxn_get_toptxn, REORDER_BUFFER_CHANGE_INVALIDATION, ReorderBufferGetChange(), ReorderBufferQueueChange(), ReorderBufferTXNByXid(), and repalloc().

Referenced by xact_decode().

ReorderBufferAddNewCommandId()

void ReorderBufferAddNewCommandId	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn,
		CommandId	cid
	)

Definition at line 3188 of file reorderbuffer.c.

{
    ReorderBufferChange *change = ReorderBufferGetChange(rb);
 
    change->data.command_id = cid;
    change->action = REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID;
 
    ReorderBufferQueueChange(rb, xid, lsn, change, false);
}

References ReorderBufferChange::action, ReorderBufferChange::command_id, ReorderBufferChange::data, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, ReorderBufferGetChange(), and ReorderBufferQueueChange().

Referenced by SnapBuildProcessNewCid().

ReorderBufferAddNewTupleCids()

void ReorderBufferAddNewTupleCids	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn,
		RelFileLocator	locator,
		ItemPointerData	tid,
		CommandId	cmin,
		CommandId	cmax,
		CommandId	combocid
	)

Definition at line 3287 of file reorderbuffer.c.

{
    ReorderBufferChange *change = ReorderBufferGetChange(rb);
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
    change->data.tuplecid.locator = locator;
    change->data.tuplecid.tid = tid;
    change->data.tuplecid.cmin = cmin;
    change->data.tuplecid.cmax = cmax;
    change->data.tuplecid.combocid = combocid;
    change->lsn = lsn;
    change->txn = txn;
    change->action = REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID;
 
    dlist_push_tail(&txn->tuplecids, &change->node);
    txn->ntuplecids++;
}

References ReorderBufferChange::action, ReorderBufferChange::data, dlist_push_tail(), ReorderBufferChange::lsn, ReorderBufferChange::node, ReorderBufferTXN::ntuplecids, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferGetChange(), ReorderBufferTXNByXid(), ReorderBufferChange::tuplecid, ReorderBufferTXN::tuplecids, and ReorderBufferChange::txn.

Referenced by SnapBuildProcessNewCid().

ReorderBufferAddSnapshot()

void ReorderBufferAddSnapshot	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn,
		Snapshot	snap
	)

Definition at line 3139 of file reorderbuffer.c.

{
    ReorderBufferChange *change = ReorderBufferGetChange(rb);
 
    change->data.snapshot = snap;
    change->action = REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT;
 
    ReorderBufferQueueChange(rb, xid, lsn, change, false);
}

References ReorderBufferChange::action, ReorderBufferChange::data, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, ReorderBufferGetChange(), ReorderBufferQueueChange(), and ReorderBufferChange::snapshot.

Referenced by SnapBuildDistributeNewCatalogSnapshot().

ReorderBufferAllocate()

ReorderBuffer* ReorderBufferAllocate

(

void

)

Definition at line 309 of file reorderbuffer.c.

{
    ReorderBuffer *buffer;
    HASHCTL     hash_ctl;
    MemoryContext new_ctx;
 
    Assert(MyReplicationSlot != NULL);
 
    /* allocate memory in own context, to have better accountability */
    new_ctx = AllocSetContextCreate(CurrentMemoryContext,
                                    "ReorderBuffer",
                                    ALLOCSET_DEFAULT_SIZES);
 
    buffer =
        (ReorderBuffer *) MemoryContextAlloc(new_ctx, sizeof(ReorderBuffer));
 
    memset(&hash_ctl, 0, sizeof(hash_ctl));
 
    buffer->context = new_ctx;
 
    buffer->change_context = SlabContextCreate(new_ctx,
                                               "Change",
                                               SLAB_DEFAULT_BLOCK_SIZE,
                                               sizeof(ReorderBufferChange));
 
    buffer->txn_context = SlabContextCreate(new_ctx,
                                            "TXN",
                                            SLAB_DEFAULT_BLOCK_SIZE,
                                            sizeof(ReorderBufferTXN));
 
    /*
     * XXX the allocation sizes used below pre-date generation context's block
     * growing code.  These values should likely be benchmarked and set to
     * more suitable values.
     */
    buffer->tup_context = GenerationContextCreate(new_ctx,
                                                  "Tuples",
                                                  SLAB_LARGE_BLOCK_SIZE,
                                                  SLAB_LARGE_BLOCK_SIZE,
                                                  SLAB_LARGE_BLOCK_SIZE);
 
    hash_ctl.keysize = sizeof(TransactionId);
    hash_ctl.entrysize = sizeof(ReorderBufferTXNByIdEnt);
    hash_ctl.hcxt = buffer->context;
 
    buffer->by_txn = hash_create("ReorderBufferByXid", 1000, &hash_ctl,
                                 HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
 
    buffer->by_txn_last_xid = InvalidTransactionId;
    buffer->by_txn_last_txn = NULL;
 
    buffer->outbuf = NULL;
    buffer->outbufsize = 0;
    buffer->size = 0;
 
    /* txn_heap is ordered by transaction size */
    buffer->txn_heap = pairingheap_allocate(ReorderBufferTXNSizeCompare, NULL);
 
    buffer->spillTxns = 0;
    buffer->spillCount = 0;
    buffer->spillBytes = 0;
    buffer->streamTxns = 0;
    buffer->streamCount = 0;
    buffer->streamBytes = 0;
    buffer->totalTxns = 0;
    buffer->totalBytes = 0;
 
    buffer->current_restart_decoding_lsn = InvalidXLogRecPtr;
 
    dlist_init(&buffer->toplevel_by_lsn);
    dlist_init(&buffer->txns_by_base_snapshot_lsn);
    dclist_init(&buffer->catchange_txns);
 
    /*
     * Ensure there's no stale data from prior uses of this slot, in case some
     * prior exit avoided calling ReorderBufferFree. Failure to do this can
     * produce duplicated txns, and it's very cheap if there's nothing there.
     */
    ReorderBufferCleanupSerializedTXNs(NameStr(MyReplicationSlot->data.name));
 
    return buffer;
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, ReorderBuffer::by_txn, ReorderBuffer::by_txn_last_txn, ReorderBuffer::by_txn_last_xid, ReorderBuffer::catchange_txns, ReorderBuffer::change_context, ReorderBuffer::context, ReorderBuffer::current_restart_decoding_lsn, CurrentMemoryContext, ReplicationSlot::data, dclist_init(), dlist_init(), HASHCTL::entrysize, GenerationContextCreate(), HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASHCTL::hcxt, InvalidTransactionId, InvalidXLogRecPtr, HASHCTL::keysize, MemoryContextAlloc(), MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReorderBuffer::outbuf, ReorderBuffer::outbufsize, pairingheap_allocate(), ReorderBufferCleanupSerializedTXNs(), ReorderBufferTXNSizeCompare(), ReorderBuffer::size, SLAB_DEFAULT_BLOCK_SIZE, SLAB_LARGE_BLOCK_SIZE, SlabContextCreate(), ReorderBuffer::spillBytes, ReorderBuffer::spillCount, ReorderBuffer::spillTxns, ReorderBuffer::streamBytes, ReorderBuffer::streamCount, ReorderBuffer::streamTxns, ReorderBuffer::toplevel_by_lsn, ReorderBuffer::totalBytes, ReorderBuffer::totalTxns, ReorderBuffer::tup_context, ReorderBuffer::txn_context, ReorderBuffer::txn_heap, and ReorderBuffer::txns_by_base_snapshot_lsn.

Referenced by StartupDecodingContext().

ReorderBufferApplyChange()

static void ReorderBufferApplyChange ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, Relation  relation, ReorderBufferChange *  change, bool  streaming  )

inlinestatic

Definition at line 1989 of file reorderbuffer.c.

{
    if (streaming)
        rb->stream_change(rb, txn, relation, change);
    else
        rb->apply_change(rb, txn, relation, change);
}

References ReorderBuffer::apply_change, and ReorderBuffer::stream_change.

Referenced by ReorderBufferProcessTXN().

ReorderBufferApplyMessage()

static void ReorderBufferApplyMessage ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, ReorderBufferChange *  change, bool  streaming  )

inlinestatic

Definition at line 2017 of file reorderbuffer.c.

{
    if (streaming)
        rb->stream_message(rb, txn, change->lsn, true,
                           change->data.msg.prefix,
                           change->data.msg.message_size,
                           change->data.msg.message);
    else
        rb->message(rb, txn, change->lsn, true,
                    change->data.msg.prefix,
                    change->data.msg.message_size,
                    change->data.msg.message);
}

References ReorderBufferChange::data, ReorderBufferChange::lsn, ReorderBuffer::message, ReorderBufferChange::msg, and ReorderBuffer::stream_message.

Referenced by ReorderBufferProcessTXN().

ReorderBufferApplyTruncate()

static void ReorderBufferApplyTruncate ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, int  nrelations, Relation *  relations, ReorderBufferChange *  change, bool  streaming  )

inlinestatic

Definition at line 2003 of file reorderbuffer.c.

{
    if (streaming)
        rb->stream_truncate(rb, txn, nrelations, relations, change);
    else
        rb->apply_truncate(rb, txn, nrelations, relations, change);
}

References ReorderBuffer::apply_truncate, and ReorderBuffer::stream_truncate.

Referenced by ReorderBufferProcessTXN().

ReorderBufferAssignChild()

void ReorderBufferAssignChild	(	ReorderBuffer *	rb,
		TransactionId	xid,
		TransactionId	subxid,
		XLogRecPtr	lsn
	)

Definition at line 1072 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
    ReorderBufferTXN *subtxn;
    bool        new_top;
    bool        new_sub;
 
    txn = ReorderBufferTXNByXid(rb, xid, true, &new_top, lsn, true);
    subtxn = ReorderBufferTXNByXid(rb, subxid, true, &new_sub, lsn, false);
 
    if (!new_sub)
    {
        if (rbtxn_is_known_subxact(subtxn))
        {
            /* already associated, nothing to do */
            return;
        }
        else
        {
            /*
             * We already saw this transaction, but initially added it to the
             * list of top-level txns.  Now that we know it's not top-level,
             * remove it from there.
             */
            dlist_delete(&subtxn->node);
        }
    }
 
    subtxn->txn_flags |= RBTXN_IS_SUBXACT;
    subtxn->toplevel_xid = xid;
    Assert(subtxn->nsubtxns == 0);
 
    /* set the reference to top-level transaction */
    subtxn->toptxn = txn;
 
    /* add to subtransaction list */
    dlist_push_tail(&txn->subtxns, &subtxn->node);
    txn->nsubtxns++;
 
    /* Possibly transfer the subtxn's snapshot to its top-level txn. */
    ReorderBufferTransferSnapToParent(txn, subtxn);
 
    /* Verify LSN-ordering invariant */
    AssertTXNLsnOrder(rb);
}

References Assert, AssertTXNLsnOrder(), dlist_delete(), dlist_push_tail(), ReorderBufferTXN::node, ReorderBufferTXN::nsubtxns, rbtxn_is_known_subxact, RBTXN_IS_SUBXACT, ReorderBufferTransferSnapToParent(), ReorderBufferTXNByXid(), ReorderBufferTXN::subtxns, ReorderBufferTXN::toplevel_xid, ReorderBufferTXN::toptxn, and ReorderBufferTXN::txn_flags.

Referenced by LogicalDecodingProcessRecord(), and ReorderBufferCommitChild().

ReorderBufferBuildTupleCidHash()

static void ReorderBufferBuildTupleCidHash ( ReorderBuffer *  rb, ReorderBufferTXN *  txn  )

static

Definition at line 1755 of file reorderbuffer.c.

{
    dlist_iter  iter;
    HASHCTL     hash_ctl;
 
    if (!rbtxn_has_catalog_changes(txn) || dlist_is_empty(&txn->tuplecids))
        return;
 
    hash_ctl.keysize = sizeof(ReorderBufferTupleCidKey);
    hash_ctl.entrysize = sizeof(ReorderBufferTupleCidEnt);
    hash_ctl.hcxt = rb->context;
 
    /*
     * create the hash with the exact number of to-be-stored tuplecids from
     * the start
     */
    txn->tuplecid_hash =
        hash_create("ReorderBufferTupleCid", txn->ntuplecids, &hash_ctl,
                    HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
 
    dlist_foreach(iter, &txn->tuplecids)
    {
        ReorderBufferTupleCidKey key;
        ReorderBufferTupleCidEnt *ent;
        bool        found;
        ReorderBufferChange *change;
 
        change = dlist_container(ReorderBufferChange, node, iter.cur);
 
        Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
 
        /* be careful about padding */
        memset(&key, 0, sizeof(ReorderBufferTupleCidKey));
 
        key.rlocator = change->data.tuplecid.locator;
 
        ItemPointerCopy(&change->data.tuplecid.tid,
                        &key.tid);
 
        ent = (ReorderBufferTupleCidEnt *)
            hash_search(txn->tuplecid_hash, &key, HASH_ENTER, &found);
        if (!found)
        {
            ent->cmin = change->data.tuplecid.cmin;
            ent->cmax = change->data.tuplecid.cmax;
            ent->combocid = change->data.tuplecid.combocid;
        }
        else
        {
            /*
             * Maybe we already saw this tuple before in this transaction, but
             * if so it must have the same cmin.
             */
            Assert(ent->cmin == change->data.tuplecid.cmin);
 
            /*
             * cmax may be initially invalid, but once set it can only grow,
             * and never become invalid again.
             */
            Assert((ent->cmax == InvalidCommandId) ||
                   ((change->data.tuplecid.cmax != InvalidCommandId) &&
                    (change->data.tuplecid.cmax > ent->cmax)));
            ent->cmax = change->data.tuplecid.cmax;
        }
    }
}

References ReorderBufferChange::action, Assert, ReorderBufferTupleCidEnt::cmax, ReorderBufferTupleCidEnt::cmin, ReorderBufferTupleCidEnt::combocid, ReorderBuffer::context, dlist_iter::cur, ReorderBufferChange::data, dlist_container, dlist_foreach, dlist_is_empty(), HASHCTL::entrysize, HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASH_ENTER, hash_search(), HASHCTL::hcxt, InvalidCommandId, ItemPointerCopy(), sort-test::key, HASHCTL::keysize, ReorderBufferTXN::ntuplecids, rbtxn_has_catalog_changes, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferChange::tuplecid, ReorderBufferTXN::tuplecid_hash, and ReorderBufferTXN::tuplecids.

Referenced by ReorderBufferProcessTXN().

ReorderBufferCanStartStreaming()

static bool ReorderBufferCanStartStreaming ( ReorderBuffer *  rb )

inlinestatic

Definition at line 4024 of file reorderbuffer.c.

{
    LogicalDecodingContext *ctx = rb->private_data;
    SnapBuild  *builder = ctx->snapshot_builder;
 
    /* We can't start streaming unless a consistent state is reached. */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_CONSISTENT)
        return false;
 
    /*
     * We can't start streaming immediately even if the streaming is enabled
     * because we previously decoded this transaction and now just are
     * restarting.
     */
    if (ReorderBufferCanStream(rb) &&
        !SnapBuildXactNeedsSkip(builder, ctx->reader->ReadRecPtr))
        return true;
 
    return false;
}

References ReorderBuffer::private_data, LogicalDecodingContext::reader, XLogReaderState::ReadRecPtr, ReorderBufferCanStream(), SNAPBUILD_CONSISTENT, SnapBuildCurrentState(), SnapBuildXactNeedsSkip(), and LogicalDecodingContext::snapshot_builder.

Referenced by ReorderBufferCheckMemoryLimit(), and ReorderBufferProcessPartialChange().

ReorderBufferCanStream()

static bool ReorderBufferCanStream ( ReorderBuffer *  rb )

inlinestatic

Definition at line 4015 of file reorderbuffer.c.

{
    LogicalDecodingContext *ctx = rb->private_data;
 
    return ctx->streaming;
}

References ReorderBuffer::private_data, and LogicalDecodingContext::streaming.

Referenced by ReorderBufferCanStartStreaming(), and ReorderBufferProcessPartialChange().

ReorderBufferChangeMemoryUpdate()

static void ReorderBufferChangeMemoryUpdate ( ReorderBuffer *  rb, ReorderBufferChange *  change, ReorderBufferTXN *  txn, bool  addition, Size  sz  )

static

Definition at line 3216 of file reorderbuffer.c.

{
    ReorderBufferTXN *toptxn;
 
    Assert(txn || change);
 
    /*
     * Ignore tuple CID changes, because those are not evicted when reaching
     * memory limit. So we just don't count them, because it might easily
     * trigger a pointless attempt to spill.
     */
    if (change && change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID)
        return;
 
    if (sz == 0)
        return;
 
    if (txn == NULL)
        txn = change->txn;
    Assert(txn != NULL);
 
    /*
     * Update the total size in top level as well. This is later used to
     * compute the decoding stats.
     */
    toptxn = rbtxn_get_toptxn(txn);
 
    if (addition)
    {
        Size        oldsize = txn->size;
 
        txn->size += sz;
        rb->size += sz;
 
        /* Update the total size in the top transaction. */
        toptxn->total_size += sz;
 
        /* Update the max-heap */
        if (oldsize != 0)
            pairingheap_remove(rb->txn_heap, &txn->txn_node);
        pairingheap_add(rb->txn_heap, &txn->txn_node);
    }
    else
    {
        Assert((rb->size >= sz) && (txn->size >= sz));
        txn->size -= sz;
        rb->size -= sz;
 
        /* Update the total size in the top transaction. */
        toptxn->total_size -= sz;
 
        /* Update the max-heap */
        pairingheap_remove(rb->txn_heap, &txn->txn_node);
        if (txn->size != 0)
            pairingheap_add(rb->txn_heap, &txn->txn_node);
    }
 
    Assert(txn->size <= rb->size);
}

References ReorderBufferChange::action, Assert, pairingheap_add(), pairingheap_remove(), rbtxn_get_toptxn, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferTXN::size, ReorderBuffer::size, ReorderBufferTXN::total_size, ReorderBufferChange::txn, ReorderBuffer::txn_heap, and ReorderBufferTXN::txn_node.

Referenced by ReorderBufferCleanupTXN(), ReorderBufferQueueChange(), ReorderBufferRestoreChange(), ReorderBufferReturnChange(), ReorderBufferSerializeTXN(), ReorderBufferToastReplace(), and ReorderBufferTruncateTXN().

ReorderBufferChangeSize()

static Size ReorderBufferChangeSize ( ReorderBufferChange *  change )

static

Definition at line 4167 of file reorderbuffer.c.

{
    Size        sz = sizeof(ReorderBufferChange);
 
    switch (change->action)
    {
            /* fall through these, they're all similar enough */
        case REORDER_BUFFER_CHANGE_INSERT:
        case REORDER_BUFFER_CHANGE_UPDATE:
        case REORDER_BUFFER_CHANGE_DELETE:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
            {
                HeapTuple   oldtup,
                            newtup;
                Size        oldlen = 0;
                Size        newlen = 0;
 
                oldtup = change->data.tp.oldtuple;
                newtup = change->data.tp.newtuple;
 
                if (oldtup)
                {
                    sz += sizeof(HeapTupleData);
                    oldlen = oldtup->t_len;
                    sz += oldlen;
                }
 
                if (newtup)
                {
                    sz += sizeof(HeapTupleData);
                    newlen = newtup->t_len;
                    sz += newlen;
                }
 
                break;
            }
        case REORDER_BUFFER_CHANGE_MESSAGE:
            {
                Size        prefix_size = strlen(change->data.msg.prefix) + 1;
 
                sz += prefix_size + change->data.msg.message_size +
                    sizeof(Size) + sizeof(Size);
 
                break;
            }
        case REORDER_BUFFER_CHANGE_INVALIDATION:
            {
                sz += sizeof(SharedInvalidationMessage) *
                    change->data.inval.ninvalidations;
                break;
            }
        case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
            {
                Snapshot    snap;
 
                snap = change->data.snapshot;
 
                sz += sizeof(SnapshotData) +
                    sizeof(TransactionId) * snap->xcnt +
                    sizeof(TransactionId) * snap->subxcnt;
 
                break;
            }
        case REORDER_BUFFER_CHANGE_TRUNCATE:
            {
                sz += sizeof(Oid) * change->data.truncate.nrelids;
 
                break;
            }
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
        case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
        case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
            /* ReorderBufferChange contains everything important */
            break;
    }
 
    return sz;
}

References ReorderBufferChange::action, ReorderBufferChange::data, ReorderBufferChange::inval, ReorderBufferChange::msg, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferChange::snapshot, SnapshotData::subxcnt, HeapTupleData::t_len, ReorderBufferChange::tp, ReorderBufferChange::truncate, and SnapshotData::xcnt.

Referenced by ReorderBufferCleanupTXN(), ReorderBufferQueueChange(), ReorderBufferRestoreChange(), ReorderBufferReturnChange(), ReorderBufferToastReplace(), and ReorderBufferTruncateTXN().

ReorderBufferCheckMemoryLimit()

static void ReorderBufferCheckMemoryLimit ( ReorderBuffer *  rb )

static

Definition at line 3632 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    /*
     * Bail out if debug_logical_replication_streaming is buffered and we
     * haven't exceeded the memory limit.
     */
    if (debug_logical_replication_streaming == DEBUG_LOGICAL_REP_STREAMING_BUFFERED &&
        rb->size < logical_decoding_work_mem * 1024L)
        return;
 
    /*
     * If debug_logical_replication_streaming is immediate, loop until there's
     * no change. Otherwise, loop until we reach under the memory limit. One
     * might think that just by evicting the largest (sub)transaction we will
     * come under the memory limit based on assumption that the selected
     * transaction is at least as large as the most recent change (which
     * caused us to go over the memory limit). However, that is not true
     * because a user can reduce the logical_decoding_work_mem to a smaller
     * value before the most recent change.
     */
    while (rb->size >= logical_decoding_work_mem * 1024L ||
           (debug_logical_replication_streaming == DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE &&
            rb->size > 0))
    {
        /*
         * Pick the largest transaction and evict it from memory by streaming,
         * if possible.  Otherwise, spill to disk.
         */
        if (ReorderBufferCanStartStreaming(rb) &&
            (txn = ReorderBufferLargestStreamableTopTXN(rb)) != NULL)
        {
            /* we know there has to be one, because the size is not zero */
            Assert(txn && rbtxn_is_toptxn(txn));
            Assert(txn->total_size > 0);
            Assert(rb->size >= txn->total_size);
 
            ReorderBufferStreamTXN(rb, txn);
        }
        else
        {
            /*
             * Pick the largest transaction (or subtransaction) and evict it
             * from memory by serializing it to disk.
             */
            txn = ReorderBufferLargestTXN(rb);
 
            /* we know there has to be one, because the size is not zero */
            Assert(txn);
            Assert(txn->size > 0);
            Assert(rb->size >= txn->size);
 
            ReorderBufferSerializeTXN(rb, txn);
        }
 
        /*
         * After eviction, the transaction should have no entries in memory,
         * and should use 0 bytes for changes.
         */
        Assert(txn->size == 0);
        Assert(txn->nentries_mem == 0);
    }
 
    /* We must be under the memory limit now. */
    Assert(rb->size < logical_decoding_work_mem * 1024L);
 
}

References Assert, DEBUG_LOGICAL_REP_STREAMING_BUFFERED, DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE, debug_logical_replication_streaming, logical_decoding_work_mem, ReorderBufferTXN::nentries_mem, rbtxn_is_toptxn, ReorderBufferCanStartStreaming(), ReorderBufferLargestStreamableTopTXN(), ReorderBufferLargestTXN(), ReorderBufferSerializeTXN(), ReorderBufferStreamTXN(), ReorderBufferTXN::size, ReorderBuffer::size, and ReorderBufferTXN::total_size.

Referenced by ReorderBufferQueueChange().

ReorderBufferCleanupSerializedTXNs()

static void ReorderBufferCleanupSerializedTXNs ( const char *  slotname )

static

Definition at line 4593 of file reorderbuffer.c.

{
    DIR        *spill_dir;
    struct dirent *spill_de;
    struct stat statbuf;
    char        path[MAXPGPATH * 2 + sizeof(PG_REPLSLOT_DIR)];
 
    sprintf(path, "%s/%s", PG_REPLSLOT_DIR, slotname);
 
    /* we're only handling directories here, skip if it's not ours */
    if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
        return;
 
    spill_dir = AllocateDir(path);
    while ((spill_de = ReadDirExtended(spill_dir, path, INFO)) != NULL)
    {
        /* only look at names that can be ours */
        if (strncmp(spill_de->d_name, "xid", 3) == 0)
        {
            snprintf(path, sizeof(path),
                     "%s/%s/%s", PG_REPLSLOT_DIR, slotname,
                     spill_de->d_name);
 
            if (unlink(path) != 0)
                ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not remove file \"%s\" during removal of %s/%s/xid*: %m",
                                path, PG_REPLSLOT_DIR, slotname)));
        }
    }
    FreeDir(spill_dir);
}

References AllocateDir(), dirent::d_name, ereport, errcode_for_file_access(), errmsg(), ERROR, FreeDir(), INFO, lstat, MAXPGPATH, PG_REPLSLOT_DIR, ReadDirExtended(), S_ISDIR, snprintf, sprintf, and stat::st_mode.

Referenced by ReorderBufferAllocate(), ReorderBufferFree(), and StartupReorderBuffer().

ReorderBufferCleanupTXN()

static void ReorderBufferCleanupTXN ( ReorderBuffer *  rb, ReorderBufferTXN *  txn  )

static

Definition at line 1508 of file reorderbuffer.c.

{
    bool        found;
    dlist_mutable_iter iter;
    Size        mem_freed = 0;
 
    /* cleanup subtransactions & their changes */
    dlist_foreach_modify(iter, &txn->subtxns)
    {
        ReorderBufferTXN *subtxn;
 
        subtxn = dlist_container(ReorderBufferTXN, node, iter.cur);
 
        /*
         * Subtransactions are always associated to the toplevel TXN, even if
         * they originally were happening inside another subtxn, so we won't
         * ever recurse more than one level deep here.
         */
        Assert(rbtxn_is_known_subxact(subtxn));
        Assert(subtxn->nsubtxns == 0);
 
        ReorderBufferCleanupTXN(rb, subtxn);
    }
 
    /* cleanup changes in the txn */
    dlist_foreach_modify(iter, &txn->changes)
    {
        ReorderBufferChange *change;
 
        change = dlist_container(ReorderBufferChange, node, iter.cur);
 
        /* Check we're not mixing changes from different transactions. */
        Assert(change->txn == txn);
 
        /*
         * Instead of updating the memory counter for individual changes, we
         * sum up the size of memory to free so we can update the memory
         * counter all together below. This saves costs of maintaining the
         * max-heap.
         */
        mem_freed += ReorderBufferChangeSize(change);
 
        ReorderBufferReturnChange(rb, change, false);
    }
 
    /* Update the memory counter */
    ReorderBufferChangeMemoryUpdate(rb, NULL, txn, false, mem_freed);
 
    /*
     * Cleanup the tuplecids we stored for decoding catalog snapshot access.
     * They are always stored in the toplevel transaction.
     */
    dlist_foreach_modify(iter, &txn->tuplecids)
    {
        ReorderBufferChange *change;
 
        change = dlist_container(ReorderBufferChange, node, iter.cur);
 
        /* Check we're not mixing changes from different transactions. */
        Assert(change->txn == txn);
        Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
 
        ReorderBufferReturnChange(rb, change, true);
    }
 
    /*
     * Cleanup the base snapshot, if set.
     */
    if (txn->base_snapshot != NULL)
    {
        SnapBuildSnapDecRefcount(txn->base_snapshot);
        dlist_delete(&txn->base_snapshot_node);
    }
 
    /*
     * Cleanup the snapshot for the last streamed run.
     */
    if (txn->snapshot_now != NULL)
    {
        Assert(rbtxn_is_streamed(txn));
        ReorderBufferFreeSnap(rb, txn->snapshot_now);
    }
 
    /*
     * Remove TXN from its containing lists.
     *
     * Note: if txn is known as subxact, we are deleting the TXN from its
     * parent's list of known subxacts; this leaves the parent's nsubxacts
     * count too high, but we don't care.  Otherwise, we are deleting the TXN
     * from the LSN-ordered list of toplevel TXNs. We remove the TXN from the
     * list of catalog modifying transactions as well.
     */
    dlist_delete(&txn->node);
    if (rbtxn_has_catalog_changes(txn))
        dclist_delete_from(&rb->catchange_txns, &txn->catchange_node);
 
    /* now remove reference from buffer */
    hash_search(rb->by_txn, &txn->xid, HASH_REMOVE, &found);
    Assert(found);
 
    /* remove entries spilled to disk */
    if (rbtxn_is_serialized(txn))
        ReorderBufferRestoreCleanup(rb, txn);
 
    /* deallocate */
    ReorderBufferReturnTXN(rb, txn);
}

References ReorderBufferChange::action, Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::base_snapshot_node, ReorderBuffer::by_txn, ReorderBufferTXN::catchange_node, ReorderBuffer::catchange_txns, ReorderBufferTXN::changes, dlist_mutable_iter::cur, dclist_delete_from(), dlist_container, dlist_delete(), dlist_foreach_modify, HASH_REMOVE, hash_search(), ReorderBufferTXN::node, ReorderBufferTXN::nsubtxns, rbtxn_has_catalog_changes, rbtxn_is_known_subxact, rbtxn_is_serialized, rbtxn_is_streamed, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferFreeSnap(), ReorderBufferRestoreCleanup(), ReorderBufferReturnChange(), ReorderBufferReturnTXN(), SnapBuildSnapDecRefcount(), ReorderBufferTXN::snapshot_now, ReorderBufferTXN::subtxns, ReorderBufferTXN::tuplecids, ReorderBufferChange::txn, and ReorderBufferTXN::xid.

Referenced by ReorderBufferAbort(), ReorderBufferAbortOld(), ReorderBufferFinishPrepared(), ReorderBufferForget(), ReorderBufferProcessTXN(), ReorderBufferReplay(), and ReorderBufferStreamCommit().

ReorderBufferCommit()

void ReorderBufferCommit	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	commit_lsn,
		XLogRecPtr	end_lsn,
		TimestampTz	commit_time,
		RepOriginId	origin_id,
		XLogRecPtr	origin_lsn
	)

Definition at line 2735 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
 
    /* unknown transaction, nothing to replay */
    if (txn == NULL)
        return;
 
    ReorderBufferReplay(txn, rb, xid, commit_lsn, end_lsn, commit_time,
                        origin_id, origin_lsn);
}

References InvalidXLogRecPtr, ReorderBufferReplay(), and ReorderBufferTXNByXid().

Referenced by DecodeCommit().

ReorderBufferCommitChild()

void ReorderBufferCommitChild	(	ReorderBuffer *	rb,
		TransactionId	xid,
		TransactionId	subxid,
		XLogRecPtr	commit_lsn,
		XLogRecPtr	end_lsn
	)

Definition at line 1192 of file reorderbuffer.c.

{
    ReorderBufferTXN *subtxn;
 
    subtxn = ReorderBufferTXNByXid(rb, subxid, false, NULL,
                                   InvalidXLogRecPtr, false);
 
    /*
     * No need to do anything if that subtxn didn't contain any changes
     */
    if (!subtxn)
        return;
 
    subtxn->final_lsn = commit_lsn;
    subtxn->end_lsn = end_lsn;
 
    /*
     * Assign this subxact as a child of the toplevel xact (no-op if already
     * done.)
     */
    ReorderBufferAssignChild(rb, xid, subxid, InvalidXLogRecPtr);
}

References ReorderBufferTXN::end_lsn, ReorderBufferTXN::final_lsn, InvalidXLogRecPtr, ReorderBufferAssignChild(), and ReorderBufferTXNByXid().

Referenced by DecodeCommit(), and DecodePrepare().

ReorderBufferCopySnap()

static Snapshot ReorderBufferCopySnap ( ReorderBuffer *  rb, Snapshot  orig_snap, ReorderBufferTXN *  txn, CommandId  cid  )

static

Definition at line 1828 of file reorderbuffer.c.

{
    Snapshot    snap;
    dlist_iter  iter;
    int         i = 0;
    Size        size;
 
    size = sizeof(SnapshotData) +
        sizeof(TransactionId) * orig_snap->xcnt +
        sizeof(TransactionId) * (txn->nsubtxns + 1);
 
    snap = MemoryContextAllocZero(rb->context, size);
    memcpy(snap, orig_snap, sizeof(SnapshotData));
 
    snap->copied = true;
    snap->active_count = 1;     /* mark as active so nobody frees it */
    snap->regd_count = 0;
    snap->xip = (TransactionId *) (snap + 1);
 
    memcpy(snap->xip, orig_snap->xip, sizeof(TransactionId) * snap->xcnt);
 
    /*
     * snap->subxip contains all txids that belong to our transaction which we
     * need to check via cmin/cmax. That's why we store the toplevel
     * transaction in there as well.
     */
    snap->subxip = snap->xip + snap->xcnt;
    snap->subxip[i++] = txn->xid;
 
    /*
     * subxcnt isn't decreased when subtransactions abort, so count manually.
     * Since it's an upper boundary it is safe to use it for the allocation
     * above.
     */
    snap->subxcnt = 1;
 
    dlist_foreach(iter, &txn->subtxns)
    {
        ReorderBufferTXN *sub_txn;
 
        sub_txn = dlist_container(ReorderBufferTXN, node, iter.cur);
        snap->subxip[i++] = sub_txn->xid;
        snap->subxcnt++;
    }
 
    /* sort so we can bsearch() later */
    qsort(snap->subxip, snap->subxcnt, sizeof(TransactionId), xidComparator);
 
    /* store the specified current CommandId */
    snap->curcid = cid;
 
    return snap;
}

References SnapshotData::active_count, ReorderBuffer::context, SnapshotData::copied, dlist_iter::cur, SnapshotData::curcid, dlist_container, dlist_foreach, i, MemoryContextAllocZero(), ReorderBufferTXN::nsubtxns, qsort, SnapshotData::regd_count, size, ReorderBufferTXN::subtxns, SnapshotData::subxcnt, SnapshotData::subxip, SnapshotData::xcnt, ReorderBufferTXN::xid, xidComparator(), and SnapshotData::xip.

Referenced by ReorderBufferProcessTXN(), ReorderBufferSaveTXNSnapshot(), and ReorderBufferStreamTXN().

ReorderBufferExecuteInvalidations()

static void ReorderBufferExecuteInvalidations ( uint32  nmsgs, SharedInvalidationMessage *  msgs  )

static

Definition at line 3383 of file reorderbuffer.c.

{
    int         i;
 
    for (i = 0; i < nmsgs; i++)
        LocalExecuteInvalidationMessage(&msgs[i]);
}

References i, and LocalExecuteInvalidationMessage().

Referenced by ReorderBufferFinishPrepared(), and ReorderBufferProcessTXN().

ReorderBufferFinishPrepared()

void ReorderBufferFinishPrepared	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	commit_lsn,
		XLogRecPtr	end_lsn,
		XLogRecPtr	two_phase_at,
		TimestampTz	commit_time,
		RepOriginId	origin_id,
		XLogRecPtr	origin_lsn,
		char *	gid,
		bool	is_commit
	)

Definition at line 2841 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
    XLogRecPtr  prepare_end_lsn;
    TimestampTz prepare_time;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, commit_lsn, false);
 
    /* unknown transaction, nothing to do */
    if (txn == NULL)
        return;
 
    /*
     * By this time the txn has the prepare record information, remember it to
     * be later used for rollback.
     */
    prepare_end_lsn = txn->end_lsn;
    prepare_time = txn->xact_time.prepare_time;
 
    /* add the gid in the txn */
    txn->gid = pstrdup(gid);
 
    /*
     * It is possible that this transaction is not decoded at prepare time
     * either because by that time we didn't have a consistent snapshot, or
     * two_phase was not enabled, or it was decoded earlier but we have
     * restarted. We only need to send the prepare if it was not decoded
     * earlier. We don't need to decode the xact for aborts if it is not done
     * already.
     */
    if ((txn->final_lsn < two_phase_at) && is_commit)
    {
        txn->txn_flags |= RBTXN_PREPARE;
 
        /*
         * The prepare info must have been updated in txn even if we skip
         * prepare.
         */
        Assert(txn->final_lsn != InvalidXLogRecPtr);
 
        /*
         * By this time the txn has the prepare record information and it is
         * important to use that so that downstream gets the accurate
         * information. If instead, we have passed commit information here
         * then downstream can behave as it has already replayed commit
         * prepared after the restart.
         */
        ReorderBufferReplay(txn, rb, xid, txn->final_lsn, txn->end_lsn,
                            txn->xact_time.prepare_time, txn->origin_id, txn->origin_lsn);
    }
 
    txn->final_lsn = commit_lsn;
    txn->end_lsn = end_lsn;
    txn->xact_time.commit_time = commit_time;
    txn->origin_id = origin_id;
    txn->origin_lsn = origin_lsn;
 
    if (is_commit)
        rb->commit_prepared(rb, txn, commit_lsn);
    else
        rb->rollback_prepared(rb, txn, prepare_end_lsn, prepare_time);
 
    /* cleanup: make sure there's no cache pollution */
    ReorderBufferExecuteInvalidations(txn->ninvalidations,
                                      txn->invalidations);
    ReorderBufferCleanupTXN(rb, txn);
}

References Assert, ReorderBuffer::commit_prepared, ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, ReorderBufferTXN::final_lsn, ReorderBufferTXN::gid, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBufferTXN::prepare_time, pstrdup(), RBTXN_PREPARE, ReorderBufferCleanupTXN(), ReorderBufferExecuteInvalidations(), ReorderBufferReplay(), ReorderBufferTXNByXid(), ReorderBuffer::rollback_prepared, ReorderBufferTXN::txn_flags, and ReorderBufferTXN::xact_time.

Referenced by DecodeAbort(), and DecodeCommit().

ReorderBufferForget()

void ReorderBufferForget	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn
	)

Definition at line 3018 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
 
    /* unknown, nothing to forget */
    if (txn == NULL)
        return;
 
    /* this transaction mustn't be streamed */
    Assert(!rbtxn_is_streamed(txn));
 
    /* cosmetic... */
    txn->final_lsn = lsn;
 
    /*
     * Process cache invalidation messages if there are any. Even if we're not
     * interested in the transaction's contents, it could have manipulated the
     * catalog and we need to update the caches according to that.
     */
    if (txn->base_snapshot != NULL && txn->ninvalidations > 0)
        ReorderBufferImmediateInvalidation(rb, txn->ninvalidations,
                                           txn->invalidations);
    else
        Assert(txn->ninvalidations == 0);
 
    /* remove potential on-disk data, and deallocate */
    ReorderBufferCleanupTXN(rb, txn);
}

References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::final_lsn, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBufferImmediateInvalidation(), and ReorderBufferTXNByXid().

Referenced by DecodeCommit().

ReorderBufferFree()

void ReorderBufferFree

(

ReorderBuffer *

rb

)

Definition at line 396 of file reorderbuffer.c.

{
    MemoryContext context = rb->context;
 
    /*
     * We free separately allocated data by entirely scrapping reorderbuffer's
     * memory context.
     */
    MemoryContextDelete(context);
 
    /* Free disk space used by unconsumed reorder buffers */
    ReorderBufferCleanupSerializedTXNs(NameStr(MyReplicationSlot->data.name));
}

References context, ReorderBuffer::context, ReplicationSlot::data, MemoryContextDelete(), MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, and ReorderBufferCleanupSerializedTXNs().

Referenced by FreeDecodingContext().

ReorderBufferFreeSnap()

static void ReorderBufferFreeSnap ( ReorderBuffer *  rb, Snapshot  snap  )

static

Definition at line 1887 of file reorderbuffer.c.

{
    if (snap->copied)
        pfree(snap);
    else
        SnapBuildSnapDecRefcount(snap);
}

References SnapshotData::copied, pfree(), and SnapBuildSnapDecRefcount().

Referenced by ReorderBufferCleanupTXN(), ReorderBufferProcessTXN(), ReorderBufferReturnChange(), and ReorderBufferStreamTXN().

ReorderBufferGetCatalogChangesXacts()

TransactionId* ReorderBufferGetCatalogChangesXacts

(

ReorderBuffer *

rb

)

Definition at line 3433 of file reorderbuffer.c.

{
    dlist_iter  iter;
    TransactionId *xids = NULL;
    size_t      xcnt = 0;
 
    /* Quick return if the list is empty */
    if (dclist_count(&rb->catchange_txns) == 0)
        return NULL;
 
    /* Initialize XID array */
    xids = (TransactionId *) palloc(sizeof(TransactionId) *
                                    dclist_count(&rb->catchange_txns));
    dclist_foreach(iter, &rb->catchange_txns)
    {
        ReorderBufferTXN *txn = dclist_container(ReorderBufferTXN,
                                                 catchange_node,
                                                 iter.cur);
 
        Assert(rbtxn_has_catalog_changes(txn));
 
        xids[xcnt++] = txn->xid;
    }
 
    qsort(xids, xcnt, sizeof(TransactionId), xidComparator);
 
    Assert(xcnt == dclist_count(&rb->catchange_txns));
    return xids;
}

References Assert, ReorderBuffer::catchange_txns, dlist_iter::cur, dclist_container, dclist_count(), dclist_foreach, palloc(), qsort, rbtxn_has_catalog_changes, ReorderBufferTXN::xid, and xidComparator().

Referenced by SnapBuildSerialize().

ReorderBufferGetChange()

ReorderBufferChange* ReorderBufferGetChange

(

ReorderBuffer *

rb

)

Definition at line 480 of file reorderbuffer.c.

{
    ReorderBufferChange *change;
 
    change = (ReorderBufferChange *)
        MemoryContextAlloc(rb->change_context, sizeof(ReorderBufferChange));
 
    memset(change, 0, sizeof(ReorderBufferChange));
    return change;
}

References ReorderBuffer::change_context, and MemoryContextAlloc().

Referenced by DecodeDelete(), DecodeInsert(), DecodeMultiInsert(), DecodeSpecConfirm(), DecodeTruncate(), DecodeUpdate(), ReorderBufferAddInvalidations(), ReorderBufferAddNewCommandId(), ReorderBufferAddNewTupleCids(), ReorderBufferAddSnapshot(), ReorderBufferQueueMessage(), and ReorderBufferRestoreChange().

ReorderBufferGetOldestTXN()

ReorderBufferTXN* ReorderBufferGetOldestTXN

(

ReorderBuffer *

rb

)

Definition at line 1017 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    AssertTXNLsnOrder(rb);
 
    if (dlist_is_empty(&rb->toplevel_by_lsn))
        return NULL;
 
    txn = dlist_head_element(ReorderBufferTXN, node, &rb->toplevel_by_lsn);
 
    Assert(!rbtxn_is_known_subxact(txn));
    Assert(txn->first_lsn != InvalidXLogRecPtr);
    return txn;
}

References Assert, AssertTXNLsnOrder(), dlist_head_element, dlist_is_empty(), ReorderBufferTXN::first_lsn, InvalidXLogRecPtr, rbtxn_is_known_subxact, and ReorderBuffer::toplevel_by_lsn.

Referenced by SnapBuildProcessRunningXacts().

ReorderBufferGetOldestXmin()

TransactionId ReorderBufferGetOldestXmin

(

ReorderBuffer *

rb

)

Definition at line 1045 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    AssertTXNLsnOrder(rb);
 
    if (dlist_is_empty(&rb->txns_by_base_snapshot_lsn))
        return InvalidTransactionId;
 
    txn = dlist_head_element(ReorderBufferTXN, base_snapshot_node,
                             &rb->txns_by_base_snapshot_lsn);
    return txn->base_snapshot->xmin;
}

References AssertTXNLsnOrder(), ReorderBufferTXN::base_snapshot, dlist_head_element, dlist_is_empty(), InvalidTransactionId, ReorderBuffer::txns_by_base_snapshot_lsn, and SnapshotData::xmin.

Referenced by SnapBuildProcessRunningXacts().

ReorderBufferGetRelids()

Oid* ReorderBufferGetRelids	(	ReorderBuffer *	rb,
		int	nrelids
	)

Definition at line 598 of file reorderbuffer.c.

{
    Oid        *relids;
    Size        alloc_len;
 
    alloc_len = sizeof(Oid) * nrelids;
 
    relids = (Oid *) MemoryContextAlloc(rb->context, alloc_len);
 
    return relids;
}

References ReorderBuffer::context, and MemoryContextAlloc().

Referenced by DecodeTruncate(), and ReorderBufferRestoreChange().

ReorderBufferGetTupleBuf()

HeapTuple ReorderBufferGetTupleBuf	(	ReorderBuffer *	rb,
		Size	tuple_len
	)

Definition at line 565 of file reorderbuffer.c.

{
    HeapTuple   tuple;
    Size        alloc_len;
 
    alloc_len = tuple_len + SizeofHeapTupleHeader;
 
    tuple = (HeapTuple) MemoryContextAlloc(rb->tup_context,
                                           HEAPTUPLESIZE + alloc_len);
    tuple->t_data = (HeapTupleHeader) ((char *) tuple + HEAPTUPLESIZE);
 
    return tuple;
}

References HEAPTUPLESIZE, MemoryContextAlloc(), SizeofHeapTupleHeader, HeapTupleData::t_data, and ReorderBuffer::tup_context.

Referenced by DecodeDelete(), DecodeInsert(), DecodeMultiInsert(), DecodeUpdate(), and ReorderBufferRestoreChange().

ReorderBufferGetTXN()

static ReorderBufferTXN * ReorderBufferGetTXN ( ReorderBuffer *  rb )

static

Definition at line 414 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = (ReorderBufferTXN *)
        MemoryContextAlloc(rb->txn_context, sizeof(ReorderBufferTXN));
 
    memset(txn, 0, sizeof(ReorderBufferTXN));
 
    dlist_init(&txn->changes);
    dlist_init(&txn->tuplecids);
    dlist_init(&txn->subtxns);
 
    /* InvalidCommandId is not zero, so set it explicitly */
    txn->command_id = InvalidCommandId;
    txn->output_plugin_private = NULL;
 
    return txn;
}

References ReorderBufferTXN::changes, ReorderBufferTXN::command_id, dlist_init(), InvalidCommandId, MemoryContextAlloc(), ReorderBufferTXN::output_plugin_private, ReorderBufferTXN::subtxns, ReorderBufferTXN::tuplecids, and ReorderBuffer::txn_context.

Referenced by ReorderBufferTXNByXid().

ReorderBufferImmediateInvalidation()

void ReorderBufferImmediateInvalidation	(	ReorderBuffer *	rb,
		uint32	ninvalidations,
		SharedInvalidationMessage *	invalidations
	)

Definition at line 3090 of file reorderbuffer.c.

{
    bool        use_subtxn = IsTransactionOrTransactionBlock();
    int         i;
 
    if (use_subtxn)
        BeginInternalSubTransaction("replay");
 
    /*
     * Force invalidations to happen outside of a valid transaction - that way
     * entries will just be marked as invalid without accessing the catalog.
     * That's advantageous because we don't need to setup the full state
     * necessary for catalog access.
     */
    if (use_subtxn)
        AbortCurrentTransaction();
 
    for (i = 0; i < ninvalidations; i++)
        LocalExecuteInvalidationMessage(&invalidations[i]);
 
    if (use_subtxn)
        RollbackAndReleaseCurrentSubTransaction();
}

References AbortCurrentTransaction(), BeginInternalSubTransaction(), i, IsTransactionOrTransactionBlock(), LocalExecuteInvalidationMessage(), and RollbackAndReleaseCurrentSubTransaction().

Referenced by ReorderBufferAbort(), ReorderBufferForget(), ReorderBufferInvalidate(), and xact_decode().

ReorderBufferInvalidate()

void ReorderBufferInvalidate	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn
	)

Definition at line 3059 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
 
    /* unknown, nothing to do */
    if (txn == NULL)
        return;
 
    /*
     * Process cache invalidation messages if there are any. Even if we're not
     * interested in the transaction's contents, it could have manipulated the
     * catalog and we need to update the caches according to that.
     */
    if (txn->base_snapshot != NULL && txn->ninvalidations > 0)
        ReorderBufferImmediateInvalidation(rb, txn->ninvalidations,
                                           txn->invalidations);
    else
        Assert(txn->ninvalidations == 0);
}

References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, ReorderBufferImmediateInvalidation(), and ReorderBufferTXNByXid().

Referenced by DecodePrepare().

ReorderBufferIterCompare()

static int ReorderBufferIterCompare ( Datum  a, Datum  b, void *  arg  )

static

Definition at line 1234 of file reorderbuffer.c.

{
    ReorderBufferIterTXNState *state = (ReorderBufferIterTXNState *) arg;
    XLogRecPtr  pos_a = state->entries[DatumGetInt32(a)].lsn;
    XLogRecPtr  pos_b = state->entries[DatumGetInt32(b)].lsn;
 
    if (pos_a < pos_b)
        return 1;
    else if (pos_a == pos_b)
        return 0;
    return -1;
}

References a, arg, b, and DatumGetInt32().

Referenced by ReorderBufferIterTXNInit().

ReorderBufferIterTXNFinish()

static void ReorderBufferIterTXNFinish ( ReorderBuffer *  rb, ReorderBufferIterTXNState *  state  )

static

Definition at line 1477 of file reorderbuffer.c.

{
    int32       off;
 
    for (off = 0; off < state->nr_txns; off++)
    {
        if (state->entries[off].file.vfd != -1)
            FileClose(state->entries[off].file.vfd);
    }
 
    /* free memory we might have "leaked" in the last *Next call */
    if (!dlist_is_empty(&state->old_change))
    {
        ReorderBufferChange *change;
 
        change = dlist_container(ReorderBufferChange, node,
                                 dlist_pop_head_node(&state->old_change));
        ReorderBufferReturnChange(rb, change, true);
        Assert(dlist_is_empty(&state->old_change));
    }
 
    binaryheap_free(state->heap);
    pfree(state);
}

References Assert, binaryheap_free(), dlist_container, dlist_is_empty(), dlist_pop_head_node(), FileClose(), pfree(), and ReorderBufferReturnChange().

Referenced by ReorderBufferProcessTXN().

ReorderBufferIterTXNInit()

static void ReorderBufferIterTXNInit ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, ReorderBufferIterTXNState *volatile *  iter_state  )

static

Definition at line 1257 of file reorderbuffer.c.

{
    Size        nr_txns = 0;
    ReorderBufferIterTXNState *state;
    dlist_iter  cur_txn_i;
    int32       off;
 
    *iter_state = NULL;
 
    /* Check ordering of changes in the toplevel transaction. */
    AssertChangeLsnOrder(txn);
 
    /*
     * Calculate the size of our heap: one element for every transaction that
     * contains changes.  (Besides the transactions already in the reorder
     * buffer, we count the one we were directly passed.)
     */
    if (txn->nentries > 0)
        nr_txns++;
 
    dlist_foreach(cur_txn_i, &txn->subtxns)
    {
        ReorderBufferTXN *cur_txn;
 
        cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);
 
        /* Check ordering of changes in this subtransaction. */
        AssertChangeLsnOrder(cur_txn);
 
        if (cur_txn->nentries > 0)
            nr_txns++;
    }
 
    /* allocate iteration state */
    state = (ReorderBufferIterTXNState *)
        MemoryContextAllocZero(rb->context,
                               sizeof(ReorderBufferIterTXNState) +
                               sizeof(ReorderBufferIterTXNEntry) * nr_txns);
 
    state->nr_txns = nr_txns;
    dlist_init(&state->old_change);
 
    for (off = 0; off < state->nr_txns; off++)
    {
        state->entries[off].file.vfd = -1;
        state->entries[off].segno = 0;
    }
 
    /* allocate heap */
    state->heap = binaryheap_allocate(state->nr_txns,
                                      ReorderBufferIterCompare,
                                      state);
 
    /* Now that the state fields are initialized, it is safe to return it. */
    *iter_state = state;
 
    /*
     * Now insert items into the binary heap, in an unordered fashion.  (We
     * will run a heap assembly step at the end; this is more efficient.)
     */
 
    off = 0;
 
    /* add toplevel transaction if it contains changes */
    if (txn->nentries > 0)
    {
        ReorderBufferChange *cur_change;
 
        if (rbtxn_is_serialized(txn))
        {
            /* serialize remaining changes */
            ReorderBufferSerializeTXN(rb, txn);
            ReorderBufferRestoreChanges(rb, txn, &state->entries[off].file,
                                        &state->entries[off].segno);
        }
 
        cur_change = dlist_head_element(ReorderBufferChange, node,
                                        &txn->changes);
 
        state->entries[off].lsn = cur_change->lsn;
        state->entries[off].change = cur_change;
        state->entries[off].txn = txn;
 
        binaryheap_add_unordered(state->heap, Int32GetDatum(off++));
    }
 
    /* add subtransactions if they contain changes */
    dlist_foreach(cur_txn_i, &txn->subtxns)
    {
        ReorderBufferTXN *cur_txn;
 
        cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);
 
        if (cur_txn->nentries > 0)
        {
            ReorderBufferChange *cur_change;
 
            if (rbtxn_is_serialized(cur_txn))
            {
                /* serialize remaining changes */
                ReorderBufferSerializeTXN(rb, cur_txn);
                ReorderBufferRestoreChanges(rb, cur_txn,
                                            &state->entries[off].file,
                                            &state->entries[off].segno);
            }
            cur_change = dlist_head_element(ReorderBufferChange, node,
                                            &cur_txn->changes);
 
            state->entries[off].lsn = cur_change->lsn;
            state->entries[off].change = cur_change;
            state->entries[off].txn = cur_txn;
 
            binaryheap_add_unordered(state->heap, Int32GetDatum(off++));
        }
    }
 
    /* assemble a valid binary heap */
    binaryheap_build(state->heap);
}

References AssertChangeLsnOrder(), binaryheap_add_unordered(), binaryheap_allocate(), binaryheap_build(), ReorderBufferTXN::changes, ReorderBuffer::context, dlist_iter::cur, dlist_container, dlist_foreach, dlist_head_element, dlist_init(), Int32GetDatum(), ReorderBufferChange::lsn, MemoryContextAllocZero(), ReorderBufferTXN::nentries, rbtxn_is_serialized, ReorderBufferIterCompare(), ReorderBufferRestoreChanges(), ReorderBufferSerializeTXN(), and ReorderBufferTXN::subtxns.

Referenced by ReorderBufferProcessTXN().

ReorderBufferIterTXNNext()

static ReorderBufferChange * ReorderBufferIterTXNNext ( ReorderBuffer *  rb, ReorderBufferIterTXNState *  state  )

static

Definition at line 1385 of file reorderbuffer.c.

{
    ReorderBufferChange *change;
    ReorderBufferIterTXNEntry *entry;
    int32       off;
 
    /* nothing there anymore */
    if (state->heap->bh_size == 0)
        return NULL;
 
    off = DatumGetInt32(binaryheap_first(state->heap));
    entry = &state->entries[off];
 
    /* free memory we might have "leaked" in the previous *Next call */
    if (!dlist_is_empty(&state->old_change))
    {
        change = dlist_container(ReorderBufferChange, node,
                                 dlist_pop_head_node(&state->old_change));
        ReorderBufferReturnChange(rb, change, true);
        Assert(dlist_is_empty(&state->old_change));
    }
 
    change = entry->change;
 
    /*
     * update heap with information about which transaction has the next
     * relevant change in LSN order
     */
 
    /* there are in-memory changes */
    if (dlist_has_next(&entry->txn->changes, &entry->change->node))
    {
        dlist_node *next = dlist_next_node(&entry->txn->changes, &change->node);
        ReorderBufferChange *next_change =
            dlist_container(ReorderBufferChange, node, next);
 
        /* txn stays the same */
        state->entries[off].lsn = next_change->lsn;
        state->entries[off].change = next_change;
 
        binaryheap_replace_first(state->heap, Int32GetDatum(off));
        return change;
    }
 
    /* try to load changes from disk */
    if (entry->txn->nentries != entry->txn->nentries_mem)
    {
        /*
         * Ugly: restoring changes will reuse *Change records, thus delete the
         * current one from the per-tx list and only free in the next call.
         */
        dlist_delete(&change->node);
        dlist_push_tail(&state->old_change, &change->node);
 
        /*
         * Update the total bytes processed by the txn for which we are
         * releasing the current set of changes and restoring the new set of
         * changes.
         */
        rb->totalBytes += entry->txn->size;
        if (ReorderBufferRestoreChanges(rb, entry->txn, &entry->file,
                                        &state->entries[off].segno))
        {
            /* successfully restored changes from disk */
            ReorderBufferChange *next_change =
                dlist_head_element(ReorderBufferChange, node,
                                   &entry->txn->changes);
 
            elog(DEBUG2, "restored %u/%u changes from disk",
                 (uint32) entry->txn->nentries_mem,
                 (uint32) entry->txn->nentries);
 
            Assert(entry->txn->nentries_mem);
            /* txn stays the same */
            state->entries[off].lsn = next_change->lsn;
            state->entries[off].change = next_change;
            binaryheap_replace_first(state->heap, Int32GetDatum(off));
 
            return change;
        }
    }
 
    /* ok, no changes there anymore, remove */
    binaryheap_remove_first(state->heap);
 
    return change;
}

References Assert, binaryheap_first(), binaryheap_remove_first(), binaryheap_replace_first(), ReorderBufferIterTXNEntry::change, ReorderBufferTXN::changes, DatumGetInt32(), DEBUG2, dlist_container, dlist_delete(), dlist_has_next(), dlist_head_element, dlist_is_empty(), dlist_next_node(), dlist_pop_head_node(), dlist_push_tail(), elog, ReorderBufferIterTXNEntry::file, Int32GetDatum(), ReorderBufferChange::lsn, ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, next, ReorderBufferChange::node, ReorderBufferRestoreChanges(), ReorderBufferReturnChange(), ReorderBufferTXN::size, ReorderBuffer::totalBytes, and ReorderBufferIterTXNEntry::txn.

Referenced by ReorderBufferProcessTXN().

ReorderBufferLargestStreamableTopTXN()

static ReorderBufferTXN* ReorderBufferLargestStreamableTopTXN ( ReorderBuffer *  rb )

static

Definition at line 3588 of file reorderbuffer.c.

{
    dlist_iter  iter;
    Size        largest_size = 0;
    ReorderBufferTXN *largest = NULL;
 
    /* Find the largest top-level transaction having a base snapshot. */
    dlist_foreach(iter, &rb->txns_by_base_snapshot_lsn)
    {
        ReorderBufferTXN *txn;
 
        txn = dlist_container(ReorderBufferTXN, base_snapshot_node, iter.cur);
 
        /* must not be a subtxn */
        Assert(!rbtxn_is_known_subxact(txn));
        /* base_snapshot must be set */
        Assert(txn->base_snapshot != NULL);
 
        if ((largest == NULL || txn->total_size > largest_size) &&
            (txn->total_size > 0) && !(rbtxn_has_partial_change(txn)) &&
            rbtxn_has_streamable_change(txn))
        {
            largest = txn;
            largest_size = txn->total_size;
        }
    }
 
    return largest;
}

References Assert, ReorderBufferTXN::base_snapshot, dlist_iter::cur, dlist_container, dlist_foreach, rbtxn_has_partial_change, rbtxn_has_streamable_change, rbtxn_is_known_subxact, ReorderBufferTXN::total_size, and ReorderBuffer::txns_by_base_snapshot_lsn.

Referenced by ReorderBufferCheckMemoryLimit().

ReorderBufferLargestTXN()

static ReorderBufferTXN* ReorderBufferLargestTXN ( ReorderBuffer *  rb )

static

Definition at line 3548 of file reorderbuffer.c.

{
    ReorderBufferTXN *largest;
 
    /* Get the largest transaction from the max-heap */
    largest = pairingheap_container(ReorderBufferTXN, txn_node,
                                    pairingheap_first(rb->txn_heap));
 
    Assert(largest);
    Assert(largest->size > 0);
    Assert(largest->size <= rb->size);
 
    return largest;
}

References Assert, pairingheap_container, pairingheap_first(), ReorderBufferTXN::size, ReorderBuffer::size, and ReorderBuffer::txn_heap.

Referenced by ReorderBufferCheckMemoryLimit().

ReorderBufferPrepare()

void ReorderBufferPrepare	(	ReorderBuffer *	rb,
		TransactionId	xid,
		char *	gid
	)

Definition at line 2804 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
 
    /* unknown transaction, nothing to replay */
    if (txn == NULL)
        return;
 
    txn->txn_flags |= RBTXN_PREPARE;
    txn->gid = pstrdup(gid);
 
    /* The prepare info must have been updated in txn by now. */
    Assert(txn->final_lsn != InvalidXLogRecPtr);
 
    ReorderBufferReplay(txn, rb, xid, txn->final_lsn, txn->end_lsn,
                        txn->xact_time.prepare_time, txn->origin_id, txn->origin_lsn);
 
    /*
     * We send the prepare for the concurrently aborted xacts so that later
     * when rollback prepared is decoded and sent, the downstream should be
     * able to rollback such a xact. See comments atop DecodePrepare.
     *
     * Note, for the concurrent_abort + streaming case a stream_prepare was
     * already sent within the ReorderBufferReplay call above.
     */
    if (txn->concurrent_abort && !rbtxn_is_streamed(txn))
        rb->prepare(rb, txn, txn->final_lsn);
}

References Assert, ReorderBufferTXN::concurrent_abort, ReorderBufferTXN::end_lsn, ReorderBufferTXN::final_lsn, ReorderBufferTXN::gid, InvalidXLogRecPtr, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBuffer::prepare, ReorderBufferTXN::prepare_time, pstrdup(), rbtxn_is_streamed, RBTXN_PREPARE, ReorderBufferReplay(), ReorderBufferTXNByXid(), ReorderBufferTXN::txn_flags, and ReorderBufferTXN::xact_time.

Referenced by DecodePrepare().

ReorderBufferProcessPartialChange()

static void ReorderBufferProcessPartialChange ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, ReorderBufferChange *  change, bool  toast_insert  )

static

Definition at line 714 of file reorderbuffer.c.

{
    ReorderBufferTXN *toptxn;
 
    /*
     * The partial changes need to be processed only while streaming
     * in-progress transactions.
     */
    if (!ReorderBufferCanStream(rb))
        return;
 
    /* Get the top transaction. */
    toptxn = rbtxn_get_toptxn(txn);
 
    /*
     * Indicate a partial change for toast inserts.  The change will be
     * considered as complete once we get the insert or update on the main
     * table and we are sure that the pending toast chunks are not required
     * anymore.
     *
     * If we allow streaming when there are pending toast chunks then such
     * chunks won't be released till the insert (multi_insert) is complete and
     * we expect the txn to have streamed all changes after streaming.  This
     * restriction is mainly to ensure the correctness of streamed
     * transactions and it doesn't seem worth uplifting such a restriction
     * just to allow this case because anyway we will stream the transaction
     * once such an insert is complete.
     */
    if (toast_insert)
        toptxn->txn_flags |= RBTXN_HAS_PARTIAL_CHANGE;
    else if (rbtxn_has_partial_change(toptxn) &&
             IsInsertOrUpdate(change->action) &&
             change->data.tp.clear_toast_afterwards)
        toptxn->txn_flags &= ~RBTXN_HAS_PARTIAL_CHANGE;
 
    /*
     * Indicate a partial change for speculative inserts.  The change will be
     * considered as complete once we get the speculative confirm or abort
     * token.
     */
    if (IsSpecInsert(change->action))
        toptxn->txn_flags |= RBTXN_HAS_PARTIAL_CHANGE;
    else if (rbtxn_has_partial_change(toptxn) &&
             IsSpecConfirmOrAbort(change->action))
        toptxn->txn_flags &= ~RBTXN_HAS_PARTIAL_CHANGE;
 
    /*
     * Stream the transaction if it is serialized before and the changes are
     * now complete in the top-level transaction.
     *
     * The reason for doing the streaming of such a transaction as soon as we
     * get the complete change for it is that previously it would have reached
     * the memory threshold and wouldn't get streamed because of incomplete
     * changes.  Delaying such transactions would increase apply lag for them.
     */
    if (ReorderBufferCanStartStreaming(rb) &&
        !(rbtxn_has_partial_change(toptxn)) &&
        rbtxn_is_serialized(txn) &&
        rbtxn_has_streamable_change(toptxn))
        ReorderBufferStreamTXN(rb, toptxn);
}

References ReorderBufferChange::action, ReorderBufferChange::data, IsInsertOrUpdate, IsSpecConfirmOrAbort, IsSpecInsert, rbtxn_get_toptxn, RBTXN_HAS_PARTIAL_CHANGE, rbtxn_has_partial_change, rbtxn_has_streamable_change, rbtxn_is_serialized, ReorderBufferCanStartStreaming(), ReorderBufferCanStream(), ReorderBufferStreamTXN(), ReorderBufferChange::tp, and ReorderBufferTXN::txn_flags.

Referenced by ReorderBufferQueueChange().

ReorderBufferProcessTXN()

static void ReorderBufferProcessTXN ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, XLogRecPtr  commit_lsn, volatile Snapshot  snapshot_now, volatile CommandId  command_id, bool  streaming  )

static

Definition at line 2104 of file reorderbuffer.c.

{
    bool        using_subtxn;
    MemoryContext ccxt = CurrentMemoryContext;
    ReorderBufferIterTXNState *volatile iterstate = NULL;
    volatile XLogRecPtr prev_lsn = InvalidXLogRecPtr;
    ReorderBufferChange *volatile specinsert = NULL;
    volatile bool stream_started = false;
    ReorderBufferTXN *volatile curtxn = NULL;
 
    /* build data to be able to lookup the CommandIds of catalog tuples */
    ReorderBufferBuildTupleCidHash(rb, txn);
 
    /* setup the initial snapshot */
    SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
 
    /*
     * Decoding needs access to syscaches et al., which in turn use
     * heavyweight locks and such. Thus we need to have enough state around to
     * keep track of those.  The easiest way is to simply use a transaction
     * internally.  That also allows us to easily enforce that nothing writes
     * to the database by checking for xid assignments.
     *
     * When we're called via the SQL SRF there's already a transaction
     * started, so start an explicit subtransaction there.
     */
    using_subtxn = IsTransactionOrTransactionBlock();
 
    PG_TRY();
    {
        ReorderBufferChange *change;
        int         changes_count = 0;  /* used to accumulate the number of
                                         * changes */
 
        if (using_subtxn)
            BeginInternalSubTransaction(streaming ? "stream" : "replay");
        else
            StartTransactionCommand();
 
        /*
         * We only need to send begin/begin-prepare for non-streamed
         * transactions.
         */
        if (!streaming)
        {
            if (rbtxn_prepared(txn))
                rb->begin_prepare(rb, txn);
            else
                rb->begin(rb, txn);
        }
 
        ReorderBufferIterTXNInit(rb, txn, &iterstate);
        while ((change = ReorderBufferIterTXNNext(rb, iterstate)) != NULL)
        {
            Relation    relation = NULL;
            Oid         reloid;
 
            CHECK_FOR_INTERRUPTS();
 
            /*
             * We can't call start stream callback before processing first
             * change.
             */
            if (prev_lsn == InvalidXLogRecPtr)
            {
                if (streaming)
                {
                    txn->origin_id = change->origin_id;
                    rb->stream_start(rb, txn, change->lsn);
                    stream_started = true;
                }
            }
 
            /*
             * Enforce correct ordering of changes, merged from multiple
             * subtransactions. The changes may have the same LSN due to
             * MULTI_INSERT xlog records.
             */
            Assert(prev_lsn == InvalidXLogRecPtr || prev_lsn <= change->lsn);
 
            prev_lsn = change->lsn;
 
            /*
             * Set the current xid to detect concurrent aborts. This is
             * required for the cases when we decode the changes before the
             * COMMIT record is processed.
             */
            if (streaming || rbtxn_prepared(change->txn))
            {
                curtxn = change->txn;
                SetupCheckXidLive(curtxn->xid);
            }
 
            switch (change->action)
            {
                case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
 
                    /*
                     * Confirmation for speculative insertion arrived. Simply
                     * use as a normal record. It'll be cleaned up at the end
                     * of INSERT processing.
                     */
                    if (specinsert == NULL)
                        elog(ERROR, "invalid ordering of speculative insertion changes");
                    Assert(specinsert->data.tp.oldtuple == NULL);
                    change = specinsert;
                    change->action = REORDER_BUFFER_CHANGE_INSERT;
 
                    /* intentionally fall through */
                case REORDER_BUFFER_CHANGE_INSERT:
                case REORDER_BUFFER_CHANGE_UPDATE:
                case REORDER_BUFFER_CHANGE_DELETE:
                    Assert(snapshot_now);
 
                    reloid = RelidByRelfilenumber(change->data.tp.rlocator.spcOid,
                                                  change->data.tp.rlocator.relNumber);
 
                    /*
                     * Mapped catalog tuple without data, emitted while
                     * catalog table was in the process of being rewritten. We
                     * can fail to look up the relfilenumber, because the
                     * relmapper has no "historic" view, in contrast to the
                     * normal catalog during decoding. Thus repeated rewrites
                     * can cause a lookup failure. That's OK because we do not
                     * decode catalog changes anyway. Normally such tuples
                     * would be skipped over below, but we can't identify
                     * whether the table should be logically logged without
                     * mapping the relfilenumber to the oid.
                     */
                    if (reloid == InvalidOid &&
                        change->data.tp.newtuple == NULL &&
                        change->data.tp.oldtuple == NULL)
                        goto change_done;
                    else if (reloid == InvalidOid)
                        elog(ERROR, "could not map filenumber \"%s\" to relation OID",
                             relpathperm(change->data.tp.rlocator,
                                         MAIN_FORKNUM));
 
                    relation = RelationIdGetRelation(reloid);
 
                    if (!RelationIsValid(relation))
                        elog(ERROR, "could not open relation with OID %u (for filenumber \"%s\")",
                             reloid,
                             relpathperm(change->data.tp.rlocator,
                                         MAIN_FORKNUM));
 
                    if (!RelationIsLogicallyLogged(relation))
                        goto change_done;
 
                    /*
                     * Ignore temporary heaps created during DDL unless the
                     * plugin has asked for them.
                     */
                    if (relation->rd_rel->relrewrite && !rb->output_rewrites)
                        goto change_done;
 
                    /*
                     * For now ignore sequence changes entirely. Most of the
                     * time they don't log changes using records we
                     * understand, so it doesn't make sense to handle the few
                     * cases we do.
                     */
                    if (relation->rd_rel->relkind == RELKIND_SEQUENCE)
                        goto change_done;
 
                    /* user-triggered change */
                    if (!IsToastRelation(relation))
                    {
                        ReorderBufferToastReplace(rb, txn, relation, change);
                        ReorderBufferApplyChange(rb, txn, relation, change,
                                                 streaming);
 
                        /*
                         * Only clear reassembled toast chunks if we're sure
                         * they're not required anymore. The creator of the
                         * tuple tells us.
                         */
                        if (change->data.tp.clear_toast_afterwards)
                            ReorderBufferToastReset(rb, txn);
                    }
                    /* we're not interested in toast deletions */
                    else if (change->action == REORDER_BUFFER_CHANGE_INSERT)
                    {
                        /*
                         * Need to reassemble the full toasted Datum in
                         * memory, to ensure the chunks don't get reused till
                         * we're done remove it from the list of this
                         * transaction's changes. Otherwise it will get
                         * freed/reused while restoring spooled data from
                         * disk.
                         */
                        Assert(change->data.tp.newtuple != NULL);
 
                        dlist_delete(&change->node);
                        ReorderBufferToastAppendChunk(rb, txn, relation,
                                                      change);
                    }
 
            change_done:
 
                    /*
                     * If speculative insertion was confirmed, the record
                     * isn't needed anymore.
                     */
                    if (specinsert != NULL)
                    {
                        ReorderBufferReturnChange(rb, specinsert, true);
                        specinsert = NULL;
                    }
 
                    if (RelationIsValid(relation))
                    {
                        RelationClose(relation);
                        relation = NULL;
                    }
                    break;
 
                case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
 
                    /*
                     * Speculative insertions are dealt with by delaying the
                     * processing of the insert until the confirmation record
                     * arrives. For that we simply unlink the record from the
                     * chain, so it does not get freed/reused while restoring
                     * spooled data from disk.
                     *
                     * This is safe in the face of concurrent catalog changes
                     * because the relevant relation can't be changed between
                     * speculative insertion and confirmation due to
                     * CheckTableNotInUse() and locking.
                     */
 
                    /* clear out a pending (and thus failed) speculation */
                    if (specinsert != NULL)
                    {
                        ReorderBufferReturnChange(rb, specinsert, true);
                        specinsert = NULL;
                    }
 
                    /* and memorize the pending insertion */
                    dlist_delete(&change->node);
                    specinsert = change;
                    break;
 
                case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
 
                    /*
                     * Abort for speculative insertion arrived. So cleanup the
                     * specinsert tuple and toast hash.
                     *
                     * Note that we get the spec abort change for each toast
                     * entry but we need to perform the cleanup only the first
                     * time we get it for the main table.
                     */
                    if (specinsert != NULL)
                    {
                        /*
                         * We must clean the toast hash before processing a
                         * completely new tuple to avoid confusion about the
                         * previous tuple's toast chunks.
                         */
                        Assert(change->data.tp.clear_toast_afterwards);
                        ReorderBufferToastReset(rb, txn);
 
                        /* We don't need this record anymore. */
                        ReorderBufferReturnChange(rb, specinsert, true);
                        specinsert = NULL;
                    }
                    break;
 
                case REORDER_BUFFER_CHANGE_TRUNCATE:
                    {
                        int         i;
                        int         nrelids = change->data.truncate.nrelids;
                        int         nrelations = 0;
                        Relation   *relations;
 
                        relations = palloc0(nrelids * sizeof(Relation));
                        for (i = 0; i < nrelids; i++)
                        {
                            Oid         relid = change->data.truncate.relids[i];
                            Relation    rel;
 
                            rel = RelationIdGetRelation(relid);
 
                            if (!RelationIsValid(rel))
                                elog(ERROR, "could not open relation with OID %u", relid);
 
                            if (!RelationIsLogicallyLogged(rel))
                                continue;
 
                            relations[nrelations++] = rel;
                        }
 
                        /* Apply the truncate. */
                        ReorderBufferApplyTruncate(rb, txn, nrelations,
                                                   relations, change,
                                                   streaming);
 
                        for (i = 0; i < nrelations; i++)
                            RelationClose(relations[i]);
 
                        break;
                    }
 
                case REORDER_BUFFER_CHANGE_MESSAGE:
                    ReorderBufferApplyMessage(rb, txn, change, streaming);
                    break;
 
                case REORDER_BUFFER_CHANGE_INVALIDATION:
                    /* Execute the invalidation messages locally */
                    ReorderBufferExecuteInvalidations(change->data.inval.ninvalidations,
                                                      change->data.inval.invalidations);
                    break;
 
                case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
                    /* get rid of the old */
                    TeardownHistoricSnapshot(false);
 
                    if (snapshot_now->copied)
                    {
                        ReorderBufferFreeSnap(rb, snapshot_now);
                        snapshot_now =
                            ReorderBufferCopySnap(rb, change->data.snapshot,
                                                  txn, command_id);
                    }
 
                    /*
                     * Restored from disk, need to be careful not to double
                     * free. We could introduce refcounting for that, but for
                     * now this seems infrequent enough not to care.
                     */
                    else if (change->data.snapshot->copied)
                    {
                        snapshot_now =
                            ReorderBufferCopySnap(rb, change->data.snapshot,
                                                  txn, command_id);
                    }
                    else
                    {
                        snapshot_now = change->data.snapshot;
                    }
 
                    /* and continue with the new one */
                    SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
                    break;
 
                case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
                    Assert(change->data.command_id != InvalidCommandId);
 
                    if (command_id < change->data.command_id)
                    {
                        command_id = change->data.command_id;
 
                        if (!snapshot_now->copied)
                        {
                            /* we don't use the global one anymore */
                            snapshot_now = ReorderBufferCopySnap(rb, snapshot_now,
                                                                 txn, command_id);
                        }
 
                        snapshot_now->curcid = command_id;
 
                        TeardownHistoricSnapshot(false);
                        SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
                    }
 
                    break;
 
                case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
                    elog(ERROR, "tuplecid value in changequeue");
                    break;
            }
 
            /*
             * It is possible that the data is not sent to downstream for a
             * long time either because the output plugin filtered it or there
             * is a DDL that generates a lot of data that is not processed by
             * the plugin. So, in such cases, the downstream can timeout. To
             * avoid that we try to send a keepalive message if required.
             * Trying to send a keepalive message after every change has some
             * overhead, but testing showed there is no noticeable overhead if
             * we do it after every ~100 changes.
             */
#define CHANGES_THRESHOLD 100
 
            if (++changes_count >= CHANGES_THRESHOLD)
            {
                rb->update_progress_txn(rb, txn, change->lsn);
                changes_count = 0;
            }
        }
 
        /* speculative insertion record must be freed by now */
        Assert(!specinsert);
 
        /* clean up the iterator */
        ReorderBufferIterTXNFinish(rb, iterstate);
        iterstate = NULL;
 
        /*
         * Update total transaction count and total bytes processed by the
         * transaction and its subtransactions. Ensure to not count the
         * streamed transaction multiple times.
         *
         * Note that the statistics computation has to be done after
         * ReorderBufferIterTXNFinish as it releases the serialized change
         * which we have already accounted in ReorderBufferIterTXNNext.
         */
        if (!rbtxn_is_streamed(txn))
            rb->totalTxns++;
 
        rb->totalBytes += txn->total_size;
 
        /*
         * Done with current changes, send the last message for this set of
         * changes depending upon streaming mode.
         */
        if (streaming)
        {
            if (stream_started)
            {
                rb->stream_stop(rb, txn, prev_lsn);
                stream_started = false;
            }
        }
        else
        {
            /*
             * Call either PREPARE (for two-phase transactions) or COMMIT (for
             * regular ones).
             */
            if (rbtxn_prepared(txn))
                rb->prepare(rb, txn, commit_lsn);
            else
                rb->commit(rb, txn, commit_lsn);
        }
 
        /* this is just a sanity check against bad output plugin behaviour */
        if (GetCurrentTransactionIdIfAny() != InvalidTransactionId)
            elog(ERROR, "output plugin used XID %u",
                 GetCurrentTransactionId());
 
        /*
         * Remember the command ID and snapshot for the next set of changes in
         * streaming mode.
         */
        if (streaming)
            ReorderBufferSaveTXNSnapshot(rb, txn, snapshot_now, command_id);
        else if (snapshot_now->copied)
            ReorderBufferFreeSnap(rb, snapshot_now);
 
        /* cleanup */
        TeardownHistoricSnapshot(false);
 
        /*
         * Aborting the current (sub-)transaction as a whole has the right
         * semantics. We want all locks acquired in here to be released, not
         * reassigned to the parent and we do not want any database access
         * have persistent effects.
         */
        AbortCurrentTransaction();
 
        /* make sure there's no cache pollution */
        ReorderBufferExecuteInvalidations(txn->ninvalidations, txn->invalidations);
 
        if (using_subtxn)
            RollbackAndReleaseCurrentSubTransaction();
 
        /*
         * We are here due to one of the four reasons: 1. Decoding an
         * in-progress txn. 2. Decoding a prepared txn. 3. Decoding of a
         * prepared txn that was (partially) streamed. 4. Decoding a committed
         * txn.
         *
         * For 1, we allow truncation of txn data by removing the changes
         * already streamed but still keeping other things like invalidations,
         * snapshot, and tuplecids. For 2 and 3, we indicate
         * ReorderBufferTruncateTXN to do more elaborate truncation of txn
         * data as the entire transaction has been decoded except for commit.
         * For 4, as the entire txn has been decoded, we can fully clean up
         * the TXN reorder buffer.
         */
        if (streaming || rbtxn_prepared(txn))
        {
            ReorderBufferTruncateTXN(rb, txn, rbtxn_prepared(txn));
            /* Reset the CheckXidAlive */
            CheckXidAlive = InvalidTransactionId;
        }
        else
            ReorderBufferCleanupTXN(rb, txn);
    }
    PG_CATCH();
    {
        MemoryContext ecxt = MemoryContextSwitchTo(ccxt);
        ErrorData  *errdata = CopyErrorData();
 
        /* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */
        if (iterstate)
            ReorderBufferIterTXNFinish(rb, iterstate);
 
        TeardownHistoricSnapshot(true);
 
        /*
         * Force cache invalidation to happen outside of a valid transaction
         * to prevent catalog access as we just caught an error.
         */
        AbortCurrentTransaction();
 
        /* make sure there's no cache pollution */
        ReorderBufferExecuteInvalidations(txn->ninvalidations,
                                          txn->invalidations);
 
        if (using_subtxn)
            RollbackAndReleaseCurrentSubTransaction();
 
        /*
         * The error code ERRCODE_TRANSACTION_ROLLBACK indicates a concurrent
         * abort of the (sub)transaction we are streaming or preparing. We
         * need to do the cleanup and return gracefully on this error, see
         * SetupCheckXidLive.
         *
         * This error code can be thrown by one of the callbacks we call
         * during decoding so we need to ensure that we return gracefully only
         * when we are sending the data in streaming mode and the streaming is
         * not finished yet or when we are sending the data out on a PREPARE
         * during a two-phase commit.
         */
        if (errdata->sqlerrcode == ERRCODE_TRANSACTION_ROLLBACK &&
            (stream_started || rbtxn_prepared(txn)))
        {
            /* curtxn must be set for streaming or prepared transactions */
            Assert(curtxn);
 
            /* Cleanup the temporary error state. */
            FlushErrorState();
            FreeErrorData(errdata);
            errdata = NULL;
            curtxn->concurrent_abort = true;
 
            /* Reset the TXN so that it is allowed to stream remaining data. */
            ReorderBufferResetTXN(rb, txn, snapshot_now,
                                  command_id, prev_lsn,
                                  specinsert);
        }
        else
        {
            ReorderBufferCleanupTXN(rb, txn);
            MemoryContextSwitchTo(ecxt);
            PG_RE_THROW();
        }
    }
    PG_END_TRY();
}

References AbortCurrentTransaction(), ReorderBufferChange::action, Assert, ReorderBuffer::begin, ReorderBuffer::begin_prepare, BeginInternalSubTransaction(), CHANGES_THRESHOLD, CHECK_FOR_INTERRUPTS, CheckXidAlive, ReorderBufferChange::command_id, ReorderBuffer::commit, ReorderBufferTXN::concurrent_abort, SnapshotData::copied, CopyErrorData(), SnapshotData::curcid, CurrentMemoryContext, ReorderBufferChange::data, data, dlist_delete(), elog, ERROR, FlushErrorState(), FreeErrorData(), GetCurrentTransactionId(), GetCurrentTransactionIdIfAny(), i, ReorderBufferChange::inval, ReorderBufferTXN::invalidations, InvalidCommandId, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, IsToastRelation(), IsTransactionOrTransactionBlock(), ReorderBufferChange::lsn, MAIN_FORKNUM, MemoryContextSwitchTo(), ReorderBufferTXN::ninvalidations, ReorderBufferChange::node, ReorderBufferChange::origin_id, ReorderBufferTXN::origin_id, ReorderBuffer::output_rewrites, palloc0(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, ReorderBuffer::prepare, rbtxn_is_streamed, rbtxn_prepared, RelationData::rd_rel, RelationClose(), RelationIdGetRelation(), RelationIsLogicallyLogged, RelationIsValid, RelidByRelfilenumber(), relpathperm, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferApplyChange(), ReorderBufferApplyMessage(), ReorderBufferApplyTruncate(), ReorderBufferBuildTupleCidHash(), ReorderBufferCleanupTXN(), ReorderBufferCopySnap(), ReorderBufferExecuteInvalidations(), ReorderBufferFreeSnap(), ReorderBufferIterTXNFinish(), ReorderBufferIterTXNInit(), ReorderBufferIterTXNNext(), ReorderBufferResetTXN(), ReorderBufferReturnChange(), ReorderBufferSaveTXNSnapshot(), ReorderBufferToastAppendChunk(), ReorderBufferToastReplace(), ReorderBufferToastReset(), ReorderBufferTruncateTXN(), RollbackAndReleaseCurrentSubTransaction(), SetupCheckXidLive(), SetupHistoricSnapshot(), ReorderBufferChange::snapshot, ErrorData::sqlerrcode, StartTransactionCommand(), ReorderBuffer::stream_start, ReorderBuffer::stream_stop, TeardownHistoricSnapshot(), ReorderBufferTXN::total_size, ReorderBuffer::totalBytes, ReorderBuffer::totalTxns, ReorderBufferChange::tp, ReorderBufferChange::truncate, ReorderBufferTXN::tuplecid_hash, ReorderBufferChange::txn, ReorderBuffer::update_progress_txn, and ReorderBufferTXN::xid.

Referenced by ReorderBufferReplay(), and ReorderBufferStreamTXN().

ReorderBufferProcessXid()

void ReorderBufferProcessXid	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn
	)

Definition at line 3126 of file reorderbuffer.c.

{
    /* many records won't have an xid assigned, centralize check here */
    if (xid != InvalidTransactionId)
        ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
}

References InvalidTransactionId, and ReorderBufferTXNByXid().

Referenced by heap2_decode(), heap_decode(), LogicalDecodingProcessRecord(), logicalmsg_decode(), standby_decode(), xact_decode(), and xlog_decode().

ReorderBufferQueueChange()

void ReorderBufferQueueChange	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn,
		ReorderBufferChange *	change,
		bool	toast_insert
	)

Definition at line 783 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
    /*
     * While streaming the previous changes we have detected that the
     * transaction is aborted.  So there is no point in collecting further
     * changes for it.
     */
    if (txn->concurrent_abort)
    {
        /*
         * We don't need to update memory accounting for this change as we
         * have not added it to the queue yet.
         */
        ReorderBufferReturnChange(rb, change, false);
        return;
    }
 
    /*
     * The changes that are sent downstream are considered streamable.  We
     * remember such transactions so that only those will later be considered
     * for streaming.
     */
    if (change->action == REORDER_BUFFER_CHANGE_INSERT ||
        change->action == REORDER_BUFFER_CHANGE_UPDATE ||
        change->action == REORDER_BUFFER_CHANGE_DELETE ||
        change->action == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT ||
        change->action == REORDER_BUFFER_CHANGE_TRUNCATE ||
        change->action == REORDER_BUFFER_CHANGE_MESSAGE)
    {
        ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn);
 
        toptxn->txn_flags |= RBTXN_HAS_STREAMABLE_CHANGE;
    }
 
    change->lsn = lsn;
    change->txn = txn;
 
    Assert(InvalidXLogRecPtr != lsn);
    dlist_push_tail(&txn->changes, &change->node);
    txn->nentries++;
    txn->nentries_mem++;
 
    /* update memory accounting information */
    ReorderBufferChangeMemoryUpdate(rb, change, NULL, true,
                                    ReorderBufferChangeSize(change));
 
    /* process partial change */
    ReorderBufferProcessPartialChange(rb, txn, change, toast_insert);
 
    /* check the memory limits and evict something if needed */
    ReorderBufferCheckMemoryLimit(rb);
}

References ReorderBufferChange::action, Assert, ReorderBufferTXN::changes, ReorderBufferTXN::concurrent_abort, dlist_push_tail(), InvalidXLogRecPtr, ReorderBufferChange::lsn, ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, ReorderBufferChange::node, rbtxn_get_toptxn, RBTXN_HAS_STREAMABLE_CHANGE, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferCheckMemoryLimit(), ReorderBufferProcessPartialChange(), ReorderBufferReturnChange(), ReorderBufferTXNByXid(), ReorderBufferChange::txn, and ReorderBufferTXN::txn_flags.

Referenced by DecodeDelete(), DecodeInsert(), DecodeMultiInsert(), DecodeSpecConfirm(), DecodeTruncate(), DecodeUpdate(), ReorderBufferAddInvalidations(), ReorderBufferAddNewCommandId(), ReorderBufferAddSnapshot(), and ReorderBufferQueueMessage().

ReorderBufferQueueMessage()

void ReorderBufferQueueMessage	(	ReorderBuffer *	rb,
		TransactionId	xid,
		Snapshot	snap,
		XLogRecPtr	lsn,
		bool	transactional,
		const char *	prefix,
		Size	message_size,
		const char *	message
	)

Definition at line 846 of file reorderbuffer.c.

{
    if (transactional)
    {
        MemoryContext oldcontext;
        ReorderBufferChange *change;
 
        Assert(xid != InvalidTransactionId);
 
        /*
         * We don't expect snapshots for transactional changes - we'll use the
         * snapshot derived later during apply (unless the change gets
         * skipped).
         */
        Assert(!snap);
 
        oldcontext = MemoryContextSwitchTo(rb->context);
 
        change = ReorderBufferGetChange(rb);
        change->action = REORDER_BUFFER_CHANGE_MESSAGE;
        change->data.msg.prefix = pstrdup(prefix);
        change->data.msg.message_size = message_size;
        change->data.msg.message = palloc(message_size);
        memcpy(change->data.msg.message, message, message_size);
 
        ReorderBufferQueueChange(rb, xid, lsn, change, false);
 
        MemoryContextSwitchTo(oldcontext);
    }
    else
    {
        ReorderBufferTXN *txn = NULL;
        volatile Snapshot snapshot_now = snap;
 
        /* Non-transactional changes require a valid snapshot. */
        Assert(snapshot_now);
 
        if (xid != InvalidTransactionId)
            txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
        /* setup snapshot to allow catalog access */
        SetupHistoricSnapshot(snapshot_now, NULL);
        PG_TRY();
        {
            rb->message(rb, txn, lsn, false, prefix, message_size, message);
 
            TeardownHistoricSnapshot(false);
        }
        PG_CATCH();
        {
            TeardownHistoricSnapshot(true);
            PG_RE_THROW();
        }
        PG_END_TRY();
    }
}

References ReorderBufferChange::action, Assert, ReorderBuffer::context, ReorderBufferChange::data, InvalidTransactionId, MemoryContextSwitchTo(), ReorderBuffer::message, ReorderBufferChange::msg, palloc(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, pstrdup(), REORDER_BUFFER_CHANGE_MESSAGE, ReorderBufferGetChange(), ReorderBufferQueueChange(), ReorderBufferTXNByXid(), SetupHistoricSnapshot(), and TeardownHistoricSnapshot().

Referenced by logicalmsg_decode().

ReorderBufferRememberPrepareInfo()

bool ReorderBufferRememberPrepareInfo	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	prepare_lsn,
		XLogRecPtr	end_lsn,
		TimestampTz	prepare_time,
		RepOriginId	origin_id,
		XLogRecPtr	origin_lsn
	)

Definition at line 2757 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false);
 
    /* unknown transaction, nothing to do */
    if (txn == NULL)
        return false;
 
    /*
     * Remember the prepare information to be later used by commit prepared in
     * case we skip doing prepare.
     */
    txn->final_lsn = prepare_lsn;
    txn->end_lsn = end_lsn;
    txn->xact_time.prepare_time = prepare_time;
    txn->origin_id = origin_id;
    txn->origin_lsn = origin_lsn;
 
    return true;
}

References ReorderBufferTXN::end_lsn, ReorderBufferTXN::final_lsn, InvalidXLogRecPtr, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBufferTXN::prepare_time, ReorderBufferTXNByXid(), and ReorderBufferTXN::xact_time.

Referenced by DecodePrepare().

ReorderBufferReplay()

static void ReorderBufferReplay ( ReorderBufferTXN *  txn, ReorderBuffer *  rb, TransactionId  xid, XLogRecPtr  commit_lsn, XLogRecPtr  end_lsn, TimestampTz  commit_time, RepOriginId  origin_id, XLogRecPtr  origin_lsn  )

static

Definition at line 2674 of file reorderbuffer.c.

{
    Snapshot    snapshot_now;
    CommandId   command_id = FirstCommandId;
 
    txn->final_lsn = commit_lsn;
    txn->end_lsn = end_lsn;
    txn->xact_time.commit_time = commit_time;
    txn->origin_id = origin_id;
    txn->origin_lsn = origin_lsn;
 
    /*
     * If the transaction was (partially) streamed, we need to commit it in a
     * 'streamed' way. That is, we first stream the remaining part of the
     * transaction, and then invoke stream_commit message.
     *
     * Called after everything (origin ID, LSN, ...) is stored in the
     * transaction to avoid passing that information directly.
     */
    if (rbtxn_is_streamed(txn))
    {
        ReorderBufferStreamCommit(rb, txn);
        return;
    }
 
    /*
     * If this transaction has no snapshot, it didn't make any changes to the
     * database, so there's nothing to decode.  Note that
     * ReorderBufferCommitChild will have transferred any snapshots from
     * subtransactions if there were any.
     */
    if (txn->base_snapshot == NULL)
    {
        Assert(txn->ninvalidations == 0);
 
        /*
         * Removing this txn before a commit might result in the computation
         * of an incorrect restart_lsn. See SnapBuildProcessRunningXacts.
         */
        if (!rbtxn_prepared(txn))
            ReorderBufferCleanupTXN(rb, txn);
        return;
    }
 
    snapshot_now = txn->base_snapshot;
 
    /* Process and send the changes to output plugin. */
    ReorderBufferProcessTXN(rb, txn, commit_lsn, snapshot_now,
                            command_id, false);
}

References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, ReorderBufferTXN::final_lsn, FirstCommandId, ReorderBufferTXN::ninvalidations, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, rbtxn_is_streamed, rbtxn_prepared, ReorderBufferCleanupTXN(), ReorderBufferProcessTXN(), ReorderBufferStreamCommit(), and ReorderBufferTXN::xact_time.

Referenced by ReorderBufferCommit(), ReorderBufferFinishPrepared(), and ReorderBufferPrepare().

ReorderBufferResetTXN()

static void ReorderBufferResetTXN ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, Snapshot  snapshot_now, CommandId  command_id, XLogRecPtr  last_lsn, ReorderBufferChange *  specinsert  )

static

Definition at line 2058 of file reorderbuffer.c.

{
    /* Discard the changes that we just streamed */
    ReorderBufferTruncateTXN(rb, txn, rbtxn_prepared(txn));
 
    /* Free all resources allocated for toast reconstruction */
    ReorderBufferToastReset(rb, txn);
 
    /* Return the spec insert change if it is not NULL */
    if (specinsert != NULL)
    {
        ReorderBufferReturnChange(rb, specinsert, true);
        specinsert = NULL;
    }
 
    /*
     * For the streaming case, stop the stream and remember the command ID and
     * snapshot for the streaming run.
     */
    if (rbtxn_is_streamed(txn))
    {
        rb->stream_stop(rb, txn, last_lsn);
        ReorderBufferSaveTXNSnapshot(rb, txn, snapshot_now, command_id);
    }
 
    /* All changes must be deallocated */
    Assert(txn->size == 0);
}

References Assert, rbtxn_is_streamed, rbtxn_prepared, ReorderBufferReturnChange(), ReorderBufferSaveTXNSnapshot(), ReorderBufferToastReset(), ReorderBufferTruncateTXN(), ReorderBufferTXN::size, and ReorderBuffer::stream_stop.

Referenced by ReorderBufferProcessTXN().

ReorderBufferRestoreChange()

static void ReorderBufferRestoreChange ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, char *  data  )

static

Definition at line 4395 of file reorderbuffer.c.

{
    ReorderBufferDiskChange *ondisk;
    ReorderBufferChange *change;
 
    ondisk = (ReorderBufferDiskChange *) data;
 
    change = ReorderBufferGetChange(rb);
 
    /* copy static part */
    memcpy(change, &ondisk->change, sizeof(ReorderBufferChange));
 
    data += sizeof(ReorderBufferDiskChange);
 
    /* restore individual stuff */
    switch (change->action)
    {
            /* fall through these, they're all similar enough */
        case REORDER_BUFFER_CHANGE_INSERT:
        case REORDER_BUFFER_CHANGE_UPDATE:
        case REORDER_BUFFER_CHANGE_DELETE:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
            if (change->data.tp.oldtuple)
            {
                uint32      tuplelen = ((HeapTuple) data)->t_len;
 
                change->data.tp.oldtuple =
                    ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader);
 
                /* restore ->tuple */
                memcpy(change->data.tp.oldtuple, data,
                       sizeof(HeapTupleData));
                data += sizeof(HeapTupleData);
 
                /* reset t_data pointer into the new tuplebuf */
                change->data.tp.oldtuple->t_data =
                    (HeapTupleHeader) ((char *) change->data.tp.oldtuple + HEAPTUPLESIZE);
 
                /* restore tuple data itself */
                memcpy(change->data.tp.oldtuple->t_data, data, tuplelen);
                data += tuplelen;
            }
 
            if (change->data.tp.newtuple)
            {
                /* here, data might not be suitably aligned! */
                uint32      tuplelen;
 
                memcpy(&tuplelen, data + offsetof(HeapTupleData, t_len),
                       sizeof(uint32));
 
                change->data.tp.newtuple =
                    ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader);
 
                /* restore ->tuple */
                memcpy(change->data.tp.newtuple, data,
                       sizeof(HeapTupleData));
                data += sizeof(HeapTupleData);
 
                /* reset t_data pointer into the new tuplebuf */
                change->data.tp.newtuple->t_data =
                    (HeapTupleHeader) ((char *) change->data.tp.newtuple + HEAPTUPLESIZE);
 
                /* restore tuple data itself */
                memcpy(change->data.tp.newtuple->t_data, data, tuplelen);
                data += tuplelen;
            }
 
            break;
        case REORDER_BUFFER_CHANGE_MESSAGE:
            {
                Size        prefix_size;
 
                /* read prefix */
                memcpy(&prefix_size, data, sizeof(Size));
                data += sizeof(Size);
                change->data.msg.prefix = MemoryContextAlloc(rb->context,
                                                             prefix_size);
                memcpy(change->data.msg.prefix, data, prefix_size);
                Assert(change->data.msg.prefix[prefix_size - 1] == '\0');
                data += prefix_size;
 
                /* read the message */
                memcpy(&change->data.msg.message_size, data, sizeof(Size));
                data += sizeof(Size);
                change->data.msg.message = MemoryContextAlloc(rb->context,
                                                              change->data.msg.message_size);
                memcpy(change->data.msg.message, data,
                       change->data.msg.message_size);
                data += change->data.msg.message_size;
 
                break;
            }
        case REORDER_BUFFER_CHANGE_INVALIDATION:
            {
                Size        inval_size = sizeof(SharedInvalidationMessage) *
                    change->data.inval.ninvalidations;
 
                change->data.inval.invalidations =
                    MemoryContextAlloc(rb->context, inval_size);
 
                /* read the message */
                memcpy(change->data.inval.invalidations, data, inval_size);
 
                break;
            }
        case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
            {
                Snapshot    oldsnap;
                Snapshot    newsnap;
                Size        size;
 
                oldsnap = (Snapshot) data;
 
                size = sizeof(SnapshotData) +
                    sizeof(TransactionId) * oldsnap->xcnt +
                    sizeof(TransactionId) * (oldsnap->subxcnt + 0);
 
                change->data.snapshot = MemoryContextAllocZero(rb->context, size);
 
                newsnap = change->data.snapshot;
 
                memcpy(newsnap, data, size);
                newsnap->xip = (TransactionId *)
                    (((char *) newsnap) + sizeof(SnapshotData));
                newsnap->subxip = newsnap->xip + newsnap->xcnt;
                newsnap->copied = true;
                break;
            }
            /* the base struct contains all the data, easy peasy */
        case REORDER_BUFFER_CHANGE_TRUNCATE:
            {
                Oid        *relids;
 
                relids = ReorderBufferGetRelids(rb,
                                                change->data.truncate.nrelids);
                memcpy(relids, data, change->data.truncate.nrelids * sizeof(Oid));
                change->data.truncate.relids = relids;
 
                break;
            }
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
        case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
        case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
            break;
    }
 
    dlist_push_tail(&txn->changes, &change->node);
    txn->nentries_mem++;
 
    /*
     * Update memory accounting for the restored change.  We need to do this
     * although we don't check the memory limit when restoring the changes in
     * this branch (we only do that when initially queueing the changes after
     * decoding), because we will release the changes later, and that will
     * update the accounting too (subtracting the size from the counters). And
     * we don't want to underflow there.
     */
    ReorderBufferChangeMemoryUpdate(rb, change, NULL, true,
                                    ReorderBufferChangeSize(change));
}

References ReorderBufferChange::action, Assert, ReorderBufferDiskChange::change, ReorderBufferTXN::changes, ReorderBuffer::context, SnapshotData::copied, ReorderBufferChange::data, data, dlist_push_tail(), HEAPTUPLESIZE, ReorderBufferChange::inval, MemoryContextAlloc(), MemoryContextAllocZero(), ReorderBufferChange::msg, ReorderBufferTXN::nentries_mem, ReorderBufferChange::node, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferGetChange(), ReorderBufferGetRelids(), ReorderBufferGetTupleBuf(), size, SizeofHeapTupleHeader, ReorderBufferChange::snapshot, SnapshotData::subxcnt, SnapshotData::subxip, ReorderBufferChange::tp, ReorderBufferChange::truncate, SnapshotData::xcnt, and SnapshotData::xip.

Referenced by ReorderBufferRestoreChanges().

ReorderBufferRestoreChanges()

static Size ReorderBufferRestoreChanges ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, TXNEntryFile *  file, XLogSegNo *  segno  )

static

Definition at line 4252 of file reorderbuffer.c.

{
    Size        restored = 0;
    XLogSegNo   last_segno;
    dlist_mutable_iter cleanup_iter;
    File       *fd = &file->vfd;
 
    Assert(txn->first_lsn != InvalidXLogRecPtr);
    Assert(txn->final_lsn != InvalidXLogRecPtr);
 
    /* free current entries, so we have memory for more */
    dlist_foreach_modify(cleanup_iter, &txn->changes)
    {
        ReorderBufferChange *cleanup =
            dlist_container(ReorderBufferChange, node, cleanup_iter.cur);
 
        dlist_delete(&cleanup->node);
        ReorderBufferReturnChange(rb, cleanup, true);
    }
    txn->nentries_mem = 0;
    Assert(dlist_is_empty(&txn->changes));
 
    XLByteToSeg(txn->final_lsn, last_segno, wal_segment_size);
 
    while (restored < max_changes_in_memory && *segno <= last_segno)
    {
        int         readBytes;
        ReorderBufferDiskChange *ondisk;
 
        CHECK_FOR_INTERRUPTS();
 
        if (*fd == -1)
        {
            char        path[MAXPGPATH];
 
            /* first time in */
            if (*segno == 0)
                XLByteToSeg(txn->first_lsn, *segno, wal_segment_size);
 
            Assert(*segno != 0 || dlist_is_empty(&txn->changes));
 
            /*
             * No need to care about TLIs here, only used during a single run,
             * so each LSN only maps to a specific WAL record.
             */
            ReorderBufferSerializedPath(path, MyReplicationSlot, txn->xid,
                                        *segno);
 
            *fd = PathNameOpenFile(path, O_RDONLY | PG_BINARY);
 
            /* No harm in resetting the offset even in case of failure */
            file->curOffset = 0;
 
            if (*fd < 0 && errno == ENOENT)
            {
                *fd = -1;
                (*segno)++;
                continue;
            }
            else if (*fd < 0)
                ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not open file \"%s\": %m",
                                path)));
        }
 
        /*
         * Read the statically sized part of a change which has information
         * about the total size. If we couldn't read a record, we're at the
         * end of this file.
         */
        ReorderBufferSerializeReserve(rb, sizeof(ReorderBufferDiskChange));
        readBytes = FileRead(file->vfd, rb->outbuf,
                             sizeof(ReorderBufferDiskChange),
                             file->curOffset, WAIT_EVENT_REORDER_BUFFER_READ);
 
        /* eof */
        if (readBytes == 0)
        {
            FileClose(*fd);
            *fd = -1;
            (*segno)++;
            continue;
        }
        else if (readBytes < 0)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read from reorderbuffer spill file: %m")));
        else if (readBytes != sizeof(ReorderBufferDiskChange))
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
                            readBytes,
                            (uint32) sizeof(ReorderBufferDiskChange))));
 
        file->curOffset += readBytes;
 
        ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
        ReorderBufferSerializeReserve(rb,
                                      sizeof(ReorderBufferDiskChange) + ondisk->size);
        ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
        readBytes = FileRead(file->vfd,
                             rb->outbuf + sizeof(ReorderBufferDiskChange),
                             ondisk->size - sizeof(ReorderBufferDiskChange),
                             file->curOffset,
                             WAIT_EVENT_REORDER_BUFFER_READ);
 
        if (readBytes < 0)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read from reorderbuffer spill file: %m")));
        else if (readBytes != ondisk->size - sizeof(ReorderBufferDiskChange))
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
                            readBytes,
                            (uint32) (ondisk->size - sizeof(ReorderBufferDiskChange)))));
 
        file->curOffset += readBytes;
 
        /*
         * ok, read a full change from disk, now restore it into proper
         * in-memory format
         */
        ReorderBufferRestoreChange(rb, txn, rb->outbuf);
        restored++;
    }
 
    return restored;
}

References Assert, ReorderBufferTXN::changes, CHECK_FOR_INTERRUPTS, cleanup(), dlist_mutable_iter::cur, TXNEntryFile::curOffset, dlist_container, dlist_delete(), dlist_foreach_modify, dlist_is_empty(), ereport, errcode_for_file_access(), errmsg(), ERROR, fd(), FileClose(), FileRead(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::first_lsn, InvalidXLogRecPtr, max_changes_in_memory, MAXPGPATH, MyReplicationSlot, ReorderBufferTXN::nentries_mem, ReorderBuffer::outbuf, PathNameOpenFile(), PG_BINARY, ReorderBufferRestoreChange(), ReorderBufferReturnChange(), ReorderBufferSerializedPath(), ReorderBufferSerializeReserve(), ReorderBufferDiskChange::size, TXNEntryFile::vfd, wal_segment_size, ReorderBufferTXN::xid, and XLByteToSeg.

Referenced by ReorderBufferIterTXNInit(), and ReorderBufferIterTXNNext().

ReorderBufferRestoreCleanup()

static void ReorderBufferRestoreCleanup ( ReorderBuffer *  rb, ReorderBufferTXN *  txn  )

static

Definition at line 4563 of file reorderbuffer.c.

{
    XLogSegNo   first;
    XLogSegNo   cur;
    XLogSegNo   last;
 
    Assert(txn->first_lsn != InvalidXLogRecPtr);
    Assert(txn->final_lsn != InvalidXLogRecPtr);
 
    XLByteToSeg(txn->first_lsn, first, wal_segment_size);
    XLByteToSeg(txn->final_lsn, last, wal_segment_size);
 
    /* iterate over all possible filenames, and delete them */
    for (cur = first; cur <= last; cur++)
    {
        char        path[MAXPGPATH];
 
        ReorderBufferSerializedPath(path, MyReplicationSlot, txn->xid, cur);
        if (unlink(path) != 0 && errno != ENOENT)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not remove file \"%s\": %m", path)));
    }
}

References Assert, cur, ereport, errcode_for_file_access(), errmsg(), ERROR, ReorderBufferTXN::final_lsn, ReorderBufferTXN::first_lsn, InvalidXLogRecPtr, MAXPGPATH, MyReplicationSlot, ReorderBufferSerializedPath(), wal_segment_size, ReorderBufferTXN::xid, and XLByteToSeg.

Referenced by ReorderBufferCleanupTXN(), and ReorderBufferTruncateTXN().

ReorderBufferReturnChange()

void ReorderBufferReturnChange	(	ReorderBuffer *	rb,
		ReorderBufferChange *	change,
		bool	upd_mem
	)

Definition at line 495 of file reorderbuffer.c.

{
    /* update memory accounting info */
    if (upd_mem)
        ReorderBufferChangeMemoryUpdate(rb, change, NULL, false,
                                        ReorderBufferChangeSize(change));
 
    /* free contained data */
    switch (change->action)
    {
        case REORDER_BUFFER_CHANGE_INSERT:
        case REORDER_BUFFER_CHANGE_UPDATE:
        case REORDER_BUFFER_CHANGE_DELETE:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
            if (change->data.tp.newtuple)
            {
                ReorderBufferReturnTupleBuf(change->data.tp.newtuple);
                change->data.tp.newtuple = NULL;
            }
 
            if (change->data.tp.oldtuple)
            {
                ReorderBufferReturnTupleBuf(change->data.tp.oldtuple);
                change->data.tp.oldtuple = NULL;
            }
            break;
        case REORDER_BUFFER_CHANGE_MESSAGE:
            if (change->data.msg.prefix != NULL)
                pfree(change->data.msg.prefix);
            change->data.msg.prefix = NULL;
            if (change->data.msg.message != NULL)
                pfree(change->data.msg.message);
            change->data.msg.message = NULL;
            break;
        case REORDER_BUFFER_CHANGE_INVALIDATION:
            if (change->data.inval.invalidations)
                pfree(change->data.inval.invalidations);
            change->data.inval.invalidations = NULL;
            break;
        case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
            if (change->data.snapshot)
            {
                ReorderBufferFreeSnap(rb, change->data.snapshot);
                change->data.snapshot = NULL;
            }
            break;
            /* no data in addition to the struct itself */
        case REORDER_BUFFER_CHANGE_TRUNCATE:
            if (change->data.truncate.relids != NULL)
            {
                ReorderBufferReturnRelids(rb, change->data.truncate.relids);
                change->data.truncate.relids = NULL;
            }
            break;
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
        case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
        case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
            break;
    }
 
    pfree(change);
}

References ReorderBufferChange::action, ReorderBufferChange::data, ReorderBufferChange::inval, ReorderBufferChange::msg, pfree(), REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferFreeSnap(), ReorderBufferReturnRelids(), ReorderBufferReturnTupleBuf(), ReorderBufferChange::snapshot, ReorderBufferChange::tp, and ReorderBufferChange::truncate.

Referenced by ReorderBufferCleanupTXN(), ReorderBufferIterTXNFinish(), ReorderBufferIterTXNNext(), ReorderBufferProcessTXN(), ReorderBufferQueueChange(), ReorderBufferResetTXN(), ReorderBufferRestoreChanges(), ReorderBufferSerializeTXN(), ReorderBufferToastReset(), and ReorderBufferTruncateTXN().

ReorderBufferReturnRelids()

void ReorderBufferReturnRelids	(	ReorderBuffer *	rb,
		Oid *	relids
	)

Definition at line 614 of file reorderbuffer.c.

{
    pfree(relids);
}

References pfree().

Referenced by ReorderBufferReturnChange().

ReorderBufferReturnTupleBuf()

void ReorderBufferReturnTupleBuf

(

HeapTuple

tuple

)

Definition at line 583 of file reorderbuffer.c.

{
    pfree(tuple);
}

References pfree().

Referenced by ReorderBufferReturnChange().

ReorderBufferReturnTXN()

static void ReorderBufferReturnTXN ( ReorderBuffer *  rb, ReorderBufferTXN *  txn  )

static

Definition at line 438 of file reorderbuffer.c.

{
    /* clean the lookup cache if we were cached (quite likely) */
    if (rb->by_txn_last_xid == txn->xid)
    {
        rb->by_txn_last_xid = InvalidTransactionId;
        rb->by_txn_last_txn = NULL;
    }
 
    /* free data that's contained */
 
    if (txn->gid != NULL)
    {
        pfree(txn->gid);
        txn->gid = NULL;
    }
 
    if (txn->tuplecid_hash != NULL)
    {
        hash_destroy(txn->tuplecid_hash);
        txn->tuplecid_hash = NULL;
    }
 
    if (txn->invalidations)
    {
        pfree(txn->invalidations);
        txn->invalidations = NULL;
    }
 
    /* Reset the toast hash */
    ReorderBufferToastReset(rb, txn);
 
    /* All changes must be deallocated */
    Assert(txn->size == 0);
 
    pfree(txn);
}

References Assert, ReorderBuffer::by_txn_last_txn, ReorderBuffer::by_txn_last_xid, ReorderBufferTXN::gid, hash_destroy(), ReorderBufferTXN::invalidations, InvalidTransactionId, pfree(), ReorderBufferToastReset(), ReorderBufferTXN::size, ReorderBufferTXN::tuplecid_hash, and ReorderBufferTXN::xid.

Referenced by ReorderBufferCleanupTXN().

ReorderBufferSaveTXNSnapshot()

static void ReorderBufferSaveTXNSnapshot ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, Snapshot  snapshot_now, CommandId  command_id  )

inlinestatic

Definition at line 2037 of file reorderbuffer.c.

{
    txn->command_id = command_id;
 
    /* Avoid copying if it's already copied. */
    if (snapshot_now->copied)
        txn->snapshot_now = snapshot_now;
    else
        txn->snapshot_now = ReorderBufferCopySnap(rb, snapshot_now,
                                                  txn, command_id);
}

References ReorderBufferTXN::command_id, SnapshotData::copied, ReorderBufferCopySnap(), and ReorderBufferTXN::snapshot_now.

Referenced by ReorderBufferProcessTXN(), and ReorderBufferResetTXN().

ReorderBufferSerializeChange()

static void ReorderBufferSerializeChange ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, int  fd, ReorderBufferChange *  change  )

static

Definition at line 3800 of file reorderbuffer.c.

{
    ReorderBufferDiskChange *ondisk;
    Size        sz = sizeof(ReorderBufferDiskChange);
 
    ReorderBufferSerializeReserve(rb, sz);
 
    ondisk = (ReorderBufferDiskChange *) rb->outbuf;
    memcpy(&ondisk->change, change, sizeof(ReorderBufferChange));
 
    switch (change->action)
    {
            /* fall through these, they're all similar enough */
        case REORDER_BUFFER_CHANGE_INSERT:
        case REORDER_BUFFER_CHANGE_UPDATE:
        case REORDER_BUFFER_CHANGE_DELETE:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
            {
                char       *data;
                HeapTuple   oldtup,
                            newtup;
                Size        oldlen = 0;
                Size        newlen = 0;
 
                oldtup = change->data.tp.oldtuple;
                newtup = change->data.tp.newtuple;
 
                if (oldtup)
                {
                    sz += sizeof(HeapTupleData);
                    oldlen = oldtup->t_len;
                    sz += oldlen;
                }
 
                if (newtup)
                {
                    sz += sizeof(HeapTupleData);
                    newlen = newtup->t_len;
                    sz += newlen;
                }
 
                /* make sure we have enough space */
                ReorderBufferSerializeReserve(rb, sz);
 
                data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
                /* might have been reallocated above */
                ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
                if (oldlen)
                {
                    memcpy(data, oldtup, sizeof(HeapTupleData));
                    data += sizeof(HeapTupleData);
 
                    memcpy(data, oldtup->t_data, oldlen);
                    data += oldlen;
                }
 
                if (newlen)
                {
                    memcpy(data, newtup, sizeof(HeapTupleData));
                    data += sizeof(HeapTupleData);
 
                    memcpy(data, newtup->t_data, newlen);
                    data += newlen;
                }
                break;
            }
        case REORDER_BUFFER_CHANGE_MESSAGE:
            {
                char       *data;
                Size        prefix_size = strlen(change->data.msg.prefix) + 1;
 
                sz += prefix_size + change->data.msg.message_size +
                    sizeof(Size) + sizeof(Size);
                ReorderBufferSerializeReserve(rb, sz);
 
                data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
 
                /* might have been reallocated above */
                ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
                /* write the prefix including the size */
                memcpy(data, &prefix_size, sizeof(Size));
                data += sizeof(Size);
                memcpy(data, change->data.msg.prefix,
                       prefix_size);
                data += prefix_size;
 
                /* write the message including the size */
                memcpy(data, &change->data.msg.message_size, sizeof(Size));
                data += sizeof(Size);
                memcpy(data, change->data.msg.message,
                       change->data.msg.message_size);
                data += change->data.msg.message_size;
 
                break;
            }
        case REORDER_BUFFER_CHANGE_INVALIDATION:
            {
                char       *data;
                Size        inval_size = sizeof(SharedInvalidationMessage) *
                    change->data.inval.ninvalidations;
 
                sz += inval_size;
 
                ReorderBufferSerializeReserve(rb, sz);
                data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
 
                /* might have been reallocated above */
                ondisk = (ReorderBufferDiskChange *) rb->outbuf;
                memcpy(data, change->data.inval.invalidations, inval_size);
                data += inval_size;
 
                break;
            }
        case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
            {
                Snapshot    snap;
                char       *data;
 
                snap = change->data.snapshot;
 
                sz += sizeof(SnapshotData) +
                    sizeof(TransactionId) * snap->xcnt +
                    sizeof(TransactionId) * snap->subxcnt;
 
                /* make sure we have enough space */
                ReorderBufferSerializeReserve(rb, sz);
                data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
                /* might have been reallocated above */
                ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
                memcpy(data, snap, sizeof(SnapshotData));
                data += sizeof(SnapshotData);
 
                if (snap->xcnt)
                {
                    memcpy(data, snap->xip,
                           sizeof(TransactionId) * snap->xcnt);
                    data += sizeof(TransactionId) * snap->xcnt;
                }
 
                if (snap->subxcnt)
                {
                    memcpy(data, snap->subxip,
                           sizeof(TransactionId) * snap->subxcnt);
                    data += sizeof(TransactionId) * snap->subxcnt;
                }
                break;
            }
        case REORDER_BUFFER_CHANGE_TRUNCATE:
            {
                Size        size;
                char       *data;
 
                /* account for the OIDs of truncated relations */
                size = sizeof(Oid) * change->data.truncate.nrelids;
                sz += size;
 
                /* make sure we have enough space */
                ReorderBufferSerializeReserve(rb, sz);
 
                data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
                /* might have been reallocated above */
                ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
                memcpy(data, change->data.truncate.relids, size);
                data += size;
 
                break;
            }
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
        case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
        case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
            /* ReorderBufferChange contains everything important */
            break;
    }
 
    ondisk->size = sz;
 
    errno = 0;
    pgstat_report_wait_start(WAIT_EVENT_REORDER_BUFFER_WRITE);
    if (write(fd, rb->outbuf, ondisk->size) != ondisk->size)
    {
        int         save_errno = errno;
 
        CloseTransientFile(fd);
 
        /* if write didn't set errno, assume problem is no disk space */
        errno = save_errno ? save_errno : ENOSPC;
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write to data file for XID %u: %m",
                        txn->xid)));
    }
    pgstat_report_wait_end();
 
    /*
     * Keep the transaction's final_lsn up to date with each change we send to
     * disk, so that ReorderBufferRestoreCleanup works correctly.  (We used to
     * only do this on commit and abort records, but that doesn't work if a
     * system crash leaves a transaction without its abort record).
     *
     * Make sure not to move it backwards.
     */
    if (txn->final_lsn < change->lsn)
        txn->final_lsn = change->lsn;
 
    Assert(ondisk->change.action == change->action);
}

References ReorderBufferChange::action, Assert, ReorderBufferDiskChange::change, CloseTransientFile(), ReorderBufferChange::data, data, ereport, errcode_for_file_access(), errmsg(), ERROR, fd(), ReorderBufferTXN::final_lsn, if(), ReorderBufferChange::inval, ReorderBufferChange::lsn, ReorderBufferChange::msg, ReorderBuffer::outbuf, pgstat_report_wait_end(), pgstat_report_wait_start(), REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, REORDER_BUFFER_CHANGE_INVALIDATION, REORDER_BUFFER_CHANGE_MESSAGE, REORDER_BUFFER_CHANGE_TRUNCATE, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferSerializeReserve(), ReorderBufferDiskChange::size, size, ReorderBufferChange::snapshot, SnapshotData::subxcnt, SnapshotData::subxip, HeapTupleData::t_data, HeapTupleData::t_len, ReorderBufferChange::tp, ReorderBufferChange::truncate, write, SnapshotData::xcnt, ReorderBufferTXN::xid, and SnapshotData::xip.

Referenced by ReorderBufferSerializeTXN().

ReorderBufferSerializedPath()

static void ReorderBufferSerializedPath ( char *  path, ReplicationSlot *  slot, TransactionId  xid, XLogSegNo  segno  )

static

Definition at line 4632 of file reorderbuffer.c.

{
    XLogRecPtr  recptr;
 
    XLogSegNoOffsetToRecPtr(segno, 0, wal_segment_size, recptr);
 
    snprintf(path, MAXPGPATH, "%s/%s/xid-%u-lsn-%X-%X.spill",
             PG_REPLSLOT_DIR,
             NameStr(MyReplicationSlot->data.name),
             xid, LSN_FORMAT_ARGS(recptr));
}

References ReplicationSlot::data, LSN_FORMAT_ARGS, MAXPGPATH, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, PG_REPLSLOT_DIR, snprintf, wal_segment_size, and XLogSegNoOffsetToRecPtr.

Referenced by ReorderBufferRestoreChanges(), ReorderBufferRestoreCleanup(), and ReorderBufferSerializeTXN().

ReorderBufferSerializeReserve()

static void ReorderBufferSerializeReserve ( ReorderBuffer *  rb, Size  sz  )

static

Definition at line 3515 of file reorderbuffer.c.

{
    if (!rb->outbufsize)
    {
        rb->outbuf = MemoryContextAlloc(rb->context, sz);
        rb->outbufsize = sz;
    }
    else if (rb->outbufsize < sz)
    {
        rb->outbuf = repalloc(rb->outbuf, sz);
        rb->outbufsize = sz;
    }
}

References ReorderBuffer::context, MemoryContextAlloc(), ReorderBuffer::outbuf, ReorderBuffer::outbufsize, and repalloc().

Referenced by ReorderBufferRestoreChanges(), and ReorderBufferSerializeChange().

ReorderBufferSerializeTXN()

static void ReorderBufferSerializeTXN ( ReorderBuffer *  rb, ReorderBufferTXN *  txn  )

static

Definition at line 3705 of file reorderbuffer.c.

{
    dlist_iter  subtxn_i;
    dlist_mutable_iter change_i;
    int         fd = -1;
    XLogSegNo   curOpenSegNo = 0;
    Size        spilled = 0;
    Size        size = txn->size;
 
    elog(DEBUG2, "spill %u changes in XID %u to disk",
         (uint32) txn->nentries_mem, txn->xid);
 
    /* do the same to all child TXs */
    dlist_foreach(subtxn_i, &txn->subtxns)
    {
        ReorderBufferTXN *subtxn;
 
        subtxn = dlist_container(ReorderBufferTXN, node, subtxn_i.cur);
        ReorderBufferSerializeTXN(rb, subtxn);
    }
 
    /* serialize changestream */
    dlist_foreach_modify(change_i, &txn->changes)
    {
        ReorderBufferChange *change;
 
        change = dlist_container(ReorderBufferChange, node, change_i.cur);
 
        /*
         * store in segment in which it belongs by start lsn, don't split over
         * multiple segments tho
         */
        if (fd == -1 ||
            !XLByteInSeg(change->lsn, curOpenSegNo, wal_segment_size))
        {
            char        path[MAXPGPATH];
 
            if (fd != -1)
                CloseTransientFile(fd);
 
            XLByteToSeg(change->lsn, curOpenSegNo, wal_segment_size);
 
            /*
             * No need to care about TLIs here, only used during a single run,
             * so each LSN only maps to a specific WAL record.
             */
            ReorderBufferSerializedPath(path, MyReplicationSlot, txn->xid,
                                        curOpenSegNo);
 
            /* open segment, create it if necessary */
            fd = OpenTransientFile(path,
                                   O_CREAT | O_WRONLY | O_APPEND | PG_BINARY);
 
            if (fd < 0)
                ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not open file \"%s\": %m", path)));
        }
 
        ReorderBufferSerializeChange(rb, txn, fd, change);
        dlist_delete(&change->node);
        ReorderBufferReturnChange(rb, change, false);
 
        spilled++;
    }
 
    /* Update the memory counter */
    ReorderBufferChangeMemoryUpdate(rb, NULL, txn, false, size);
 
    /* update the statistics iff we have spilled anything */
    if (spilled)
    {
        rb->spillCount += 1;
        rb->spillBytes += size;
 
        /* don't consider already serialized transactions */
        rb->spillTxns += (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1;
 
        /* update the decoding stats */
        UpdateDecodingStats((LogicalDecodingContext *) rb->private_data);
    }
 
    Assert(spilled == txn->nentries_mem);
    Assert(dlist_is_empty(&txn->changes));
    txn->nentries_mem = 0;
    txn->txn_flags |= RBTXN_IS_SERIALIZED;
 
    if (fd != -1)
        CloseTransientFile(fd);
}

References Assert, ReorderBufferTXN::changes, CloseTransientFile(), dlist_iter::cur, dlist_mutable_iter::cur, DEBUG2, dlist_container, dlist_delete(), dlist_foreach, dlist_foreach_modify, dlist_is_empty(), elog, ereport, errcode_for_file_access(), errmsg(), ERROR, fd(), ReorderBufferChange::lsn, MAXPGPATH, MyReplicationSlot, ReorderBufferTXN::nentries_mem, ReorderBufferChange::node, OpenTransientFile(), PG_BINARY, ReorderBuffer::private_data, RBTXN_IS_SERIALIZED, rbtxn_is_serialized, rbtxn_is_serialized_clear, ReorderBufferChangeMemoryUpdate(), ReorderBufferReturnChange(), ReorderBufferSerializeChange(), ReorderBufferSerializedPath(), size, ReorderBufferTXN::size, ReorderBuffer::spillBytes, ReorderBuffer::spillCount, ReorderBuffer::spillTxns, ReorderBufferTXN::subtxns, ReorderBufferTXN::txn_flags, UpdateDecodingStats(), wal_segment_size, ReorderBufferTXN::xid, XLByteInSeg, and XLByteToSeg.

Referenced by ReorderBufferCheckMemoryLimit(), and ReorderBufferIterTXNInit().

ReorderBufferSetBaseSnapshot()

void ReorderBufferSetBaseSnapshot	(	ReorderBuffer *	rb,
		TransactionId	xid,
		XLogRecPtr	lsn,
		Snapshot	snap
	)

Definition at line 3157 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
    bool        is_new;
 
    Assert(snap != NULL);
 
    /*
     * Fetch the transaction to operate on.  If we know it's a subtransaction,
     * operate on its top-level transaction instead.
     */
    txn = ReorderBufferTXNByXid(rb, xid, true, &is_new, lsn, true);
    if (rbtxn_is_known_subxact(txn))
        txn = ReorderBufferTXNByXid(rb, txn->toplevel_xid, false,
                                    NULL, InvalidXLogRecPtr, false);
    Assert(txn->base_snapshot == NULL);
 
    txn->