#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:

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 5082 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 991 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 920 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 5184 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 2931 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 2976 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 3329 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, ReorderBufferChange::invalidations, ReorderBufferTXN::invalidations, MemoryContextSwitchTo(), ReorderBufferChange::ninvalidations, 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 3193 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 3292 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::cmax, ReorderBufferChange::cmin, ReorderBufferChange::combocid, ReorderBufferChange::data, dlist_push_tail(), ReorderBufferChange::locator, ReorderBufferChange::lsn, ReorderBufferChange::node, ReorderBufferTXN::ntuplecids, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferGetChange(), ReorderBufferTXNByXid(), ReorderBufferChange::tid, ReorderBufferChange::tuplecid, ReorderBufferTXN::tuplecids, and ReorderBufferChange::txn.

Referenced by SnapBuildProcessNewCid().

◆ ReorderBufferAddSnapshot()

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

Definition at line 3144 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));
 
    /*
     * To minimize memory fragmentation caused by long-running transactions
     * with changes spanning multiple memory blocks, we use a single
     * fixed-size memory block for decoded tuple storage. The performance
     * testing showed that the default memory block size maintains logical
     * decoding performance without causing fragmentation due to concurrent
     * transactions. One might think that we can use the max size as
     * SLAB_LARGE_BLOCK_SIZE but the test also showed it doesn't help resolve
     * the memory fragmentation.
     */
    buffer->tup_context = GenerationContextCreate(new_ctx,
                                                  "Tuples",
                                                  SLAB_DEFAULT_BLOCK_SIZE,
                                                  SLAB_DEFAULT_BLOCK_SIZE,
                                                  SLAB_DEFAULT_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, 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 1994 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 2022 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, ReorderBufferChange::message, ReorderBuffer::message, ReorderBufferChange::message_size, ReorderBufferChange::msg, ReorderBufferChange::prefix, 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 2008 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 1077 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 1760 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;
        }
    }
}

Referenced by ReorderBufferProcessTXN().

◆ ReorderBufferCanStartStreaming()

static bool ReorderBufferCanStartStreaming ( ReorderBuffer * rb )

inlinestatic

Definition at line 4029 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 4020 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 3221 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 4172 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;
}

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

◆ ReorderBufferCheckMemoryLimit()

static void ReorderBufferCheckMemoryLimit ( ReorderBuffer * rb )

static

Definition at line 3637 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 4598 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 1513 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);
}

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

◆ ReorderBufferCommit()

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

Definition at line 2740 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 1197 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 1833 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;
 
    /*
     * txn->nsubtxns 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 3388 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 2846 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);
}

Referenced by DecodeAbort(), and DecodeCommit().

◆ ReorderBufferForget()

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

Definition at line 3023 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 401 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 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 1892 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 3438 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 485 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 1022 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 1050 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 603 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 570 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 419 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 3095 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 3064 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 1239 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 1482 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 1262 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 1390 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 3593 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 3553 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 2809 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 719 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::clear_toast_afterwards, 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 2109 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();
}

Referenced by ReorderBufferReplay(), and ReorderBufferStreamTXN().

◆ ReorderBufferProcessXid()

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

Definition at line 3131 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 788 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 851 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(), ReorderBufferChange::message, ReorderBuffer::message, ReorderBufferChange::message_size, ReorderBufferChange::msg, palloc(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, ReorderBufferChange::prefix, 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 2762 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 2679 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 2063 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 4400 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));
}

Referenced by ReorderBufferRestoreChanges().

◆ ReorderBufferRestoreChanges()

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

static

Definition at line 4257 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 4568 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 500 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);
}

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

◆ ReorderBufferReturnRelids()

void ReorderBufferReturnRelids	(	ReorderBuffer *	rb,
		Oid *	relids
	)

Definition at line 619 of file reorderbuffer.c.

{
    pfree(relids);
}

References pfree().

Referenced by ReorderBufferReturnChange().

◆ ReorderBufferReturnTupleBuf()

void ReorderBufferReturnTupleBuf ( HeapTuple tuple )

Definition at line 588 of file reorderbuffer.c.

{
    pfree(tuple);
}

References pfree().

Referenced by ReorderBufferReturnChange().

◆ ReorderBufferReturnTXN()

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

static

Definition at line 443 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 2042 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 3805 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);
}

Referenced by ReorderBufferSerializeTXN().

◆ ReorderBufferSerializedPath()

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

static

Definition at line 4637 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 3520 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 3710 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(), ReorderBufferSerializeTXN(), 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(), ReorderBufferIterTXNInit(), and ReorderBufferSerializeTXN().

◆ ReorderBufferSetBaseSnapshot()

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

Definition at line 3162 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->base_snapshot = snap;
    txn->base_snapshot_lsn = lsn;
    dlist_push_tail(&rb->txns_by_base_snapshot_lsn, &txn->base_snapshot_node);
 
    AssertTXNLsnOrder(rb);
}

References Assert, AssertTXNLsnOrder(), ReorderBufferTXN::base_snapshot, ReorderBufferTXN::base_snapshot_lsn, ReorderBufferTXN::base_snapshot_node, dlist_push_tail(), InvalidXLogRecPtr, rbtxn_is_known_subxact, ReorderBufferTXNByXid(), ReorderBufferTXN::toplevel_xid, and ReorderBuffer::txns_by_base_snapshot_lsn.

Referenced by SnapBuildCommitTxn(), and SnapBuildProcessChange().

◆ ReorderBufferSetRestartPoint()

void ReorderBufferSetRestartPoint	(	ReorderBuffer *	rb,
		XLogRecPtr	ptr
	)

Definition at line 1065 of file reorderbuffer.c.

{
    rb->current_restart_decoding_lsn = ptr;
}

References ReorderBuffer::current_restart_decoding_lsn.

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

◆ ReorderBufferSkipPrepare()

void ReorderBufferSkipPrepare	(	ReorderBuffer *	rb,
		TransactionId	xid
	)

Definition at line 2790 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false);
 
    /* unknown transaction, nothing to do */
    if (txn == NULL)
        return;
 
    txn->txn_flags |= RBTXN_SKIPPED_PREPARE;
}

References InvalidXLogRecPtr, RBTXN_SKIPPED_PREPARE, ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.

Referenced by DecodePrepare().

◆ ReorderBufferStreamCommit()

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

static

Definition at line 1907 of file reorderbuffer.c.

{
    /* we should only call this for previously streamed transactions */
    Assert(rbtxn_is_streamed(txn));
 
    ReorderBufferStreamTXN(rb, txn);
 
    if (rbtxn_prepared(txn))
    {
        /*
         * Note, we send stream prepare even if a concurrent abort is
         * detected. See DecodePrepare for more information.
         */
        rb->stream_prepare(rb, txn, txn->final_lsn);
 
        /*
         * This is a PREPARED transaction, part of a two-phase commit. The
         * full cleanup will happen as part of the COMMIT PREPAREDs, so now
         * just truncate txn by removing changes and tuplecids.
         */
        ReorderBufferTruncateTXN(rb, txn, true);
        /* Reset the CheckXidAlive */
        CheckXidAlive = InvalidTransactionId;
    }
    else
    {
        rb->stream_commit(rb, txn, txn->final_lsn);
        ReorderBufferCleanupTXN(rb, txn);
    }
}

References Assert, CheckXidAlive, ReorderBufferTXN::final_lsn, InvalidTransactionId, rbtxn_is_streamed, rbtxn_prepared, ReorderBufferCleanupTXN(), ReorderBufferStreamTXN(), ReorderBufferTruncateTXN(), ReorderBuffer::stream_commit, and ReorderBuffer::stream_prepare.

Referenced by ReorderBufferReplay().

◆ ReorderBufferStreamTXN()

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

static

Definition at line 4055 of file reorderbuffer.c.

{
    Snapshot    snapshot_now;
    CommandId   command_id;
    Size        stream_bytes;
    bool        txn_is_streamed;
 
    /* We can never reach here for a subtransaction. */
    Assert(rbtxn_is_toptxn(txn));
 
    /*
     * We can't make any assumptions about base snapshot here, similar to what
     * ReorderBufferCommit() does. That relies on base_snapshot getting
     * transferred from subxact in ReorderBufferCommitChild(), but that was
     * not yet called as the transaction is in-progress.
     *
     * So just walk the subxacts and use the same logic here. But we only need
     * to do that once, when the transaction is streamed for the first time.
     * After that we need to reuse the snapshot from the previous run.
     *
     * Unlike DecodeCommit which adds xids of all the subtransactions in
     * snapshot's xip array via SnapBuildCommitTxn, we can't do that here but
     * we do add them to subxip array instead via ReorderBufferCopySnap. This
     * allows the catalog changes made in subtransactions decoded till now to
     * be visible.
     */
    if (txn->snapshot_now == NULL)
    {
        dlist_iter  subxact_i;
 
        /* make sure this transaction is streamed for the first time */
        Assert(!rbtxn_is_streamed(txn));
 
        /* at the beginning we should have invalid command ID */
        Assert(txn->command_id == InvalidCommandId);
 
        dlist_foreach(subxact_i, &txn->subtxns)
        {
            ReorderBufferTXN *subtxn;
 
            subtxn = dlist_container(ReorderBufferTXN, node, subxact_i.cur);
            ReorderBufferTransferSnapToParent(txn, subtxn);
        }
 
        /*
         * If this transaction has no snapshot, it didn't make any changes to
         * the database till now, so there's nothing to decode.
         */
        if (txn->base_snapshot == NULL)
        {
            Assert(txn->ninvalidations == 0);
            return;
        }
 
        command_id = FirstCommandId;
        snapshot_now = ReorderBufferCopySnap(rb, txn->base_snapshot,
                                             txn, command_id);
    }
    else
    {
        /* the transaction must have been already streamed */
        Assert(rbtxn_is_streamed(txn));
 
        /*
         * Nah, we already have snapshot from the previous streaming run. We
         * assume new subxacts can't move the LSN backwards, and so can't beat
         * the LSN condition in the previous branch (so no need to walk
         * through subxacts again). In fact, we must not do that as we may be
         * using snapshot half-way through the subxact.
         */
        command_id = txn->command_id;
 
        /*
         * We can't use txn->snapshot_now directly because after the last
         * streaming run, we might have got some new sub-transactions. So we
         * need to add them to the snapshot.
         */
        snapshot_now = ReorderBufferCopySnap(rb, txn->snapshot_now,
                                             txn, command_id);
 
        /* Free the previously copied snapshot. */
        Assert(txn->snapshot_now->copied);
        ReorderBufferFreeSnap(rb, txn->snapshot_now);
        txn->snapshot_now = NULL;
    }
 
    /*
     * Remember this information to be used later to update stats. We can't
     * update the stats here as an error while processing the changes would
     * lead to the accumulation of stats even though we haven't streamed all
     * the changes.
     */
    txn_is_streamed = rbtxn_is_streamed(txn);
    stream_bytes = txn->total_size;
 
    /* Process and send the changes to output plugin. */
    ReorderBufferProcessTXN(rb, txn, InvalidXLogRecPtr, snapshot_now,
                            command_id, true);
 
    rb->streamCount += 1;
    rb->streamBytes += stream_bytes;
 
    /* Don't consider already streamed transaction. */
    rb->streamTxns += (txn_is_streamed) ? 0 : 1;
 
    /* update the decoding stats */
    UpdateDecodingStats((LogicalDecodingContext *) rb->private_data);
 
    Assert(dlist_is_empty(&txn->changes));
    Assert(txn->nentries == 0);
    Assert(txn->nentries_mem == 0);
}

References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::changes, ReorderBufferTXN::command_id, SnapshotData::copied, dlist_iter::cur, dlist_container, dlist_foreach, dlist_is_empty(), FirstCommandId, InvalidCommandId, InvalidXLogRecPtr, ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, ReorderBufferTXN::ninvalidations, ReorderBuffer::private_data, rbtxn_is_streamed, rbtxn_is_toptxn, ReorderBufferCopySnap(), ReorderBufferFreeSnap(), ReorderBufferProcessTXN(), ReorderBufferTransferSnapToParent(), ReorderBufferTXN::snapshot_now,

Data Structures

Macros

Typedefs

Functions

Variables

Macro Definition Documentation

◆ CHANGES_THRESHOLD

◆ IsInsertOrUpdate

◆ IsSpecConfirmOrAbort

◆ IsSpecInsert

Typedef Documentation

◆ ReorderBufferDiskChange

◆ ReorderBufferIterTXNEntry

◆ ReorderBufferIterTXNState

◆ ReorderBufferToastEnt

◆ ReorderBufferTupleCidEnt

◆ ReorderBufferTupleCidKey

◆ ReorderBufferTXNByIdEnt

◆ RewriteMappingFile

◆ TXNEntryFile

Function Documentation

◆ ApplyLogicalMappingFile()

◆ AssertChangeLsnOrder()

◆ AssertTXNLsnOrder()

◆ file_sort_by_lsn()

◆ ReorderBufferAbort()

◆ ReorderBufferAbortOld()

◆ ReorderBufferAddInvalidations()

◆ ReorderBufferAddNewCommandId()

◆ ReorderBufferAddNewTupleCids()

◆ ReorderBufferAddSnapshot()

◆ ReorderBufferAllocate()

◆ ReorderBufferApplyChange()

◆ ReorderBufferApplyMessage()

◆ ReorderBufferApplyTruncate()

◆ ReorderBufferAssignChild()

◆ ReorderBufferBuildTupleCidHash()

◆ ReorderBufferCanStartStreaming()

◆ ReorderBufferCanStream()

◆ ReorderBufferChangeMemoryUpdate()

◆ ReorderBufferChangeSize()

◆ ReorderBufferCheckMemoryLimit()

◆ ReorderBufferCleanupSerializedTXNs()

◆ ReorderBufferCleanupTXN()

◆ ReorderBufferCommit()

◆ ReorderBufferCommitChild()

◆ ReorderBufferCopySnap()

◆ ReorderBufferExecuteInvalidations()

◆ ReorderBufferFinishPrepared()

◆ ReorderBufferForget()

◆ ReorderBufferFree()

◆ ReorderBufferFreeSnap()

◆ ReorderBufferGetCatalogChangesXacts()

◆ ReorderBufferGetChange()

◆ ReorderBufferGetOldestTXN()

◆ ReorderBufferGetOldestXmin()

◆ ReorderBufferGetRelids()

◆ ReorderBufferGetTupleBuf()

◆ ReorderBufferGetTXN()

◆ ReorderBufferImmediateInvalidation()

◆ ReorderBufferInvalidate()

◆ ReorderBufferIterCompare()

◆ ReorderBufferIterTXNFinish()

◆ ReorderBufferIterTXNInit()

◆ ReorderBufferIterTXNNext()

◆ ReorderBufferLargestStreamableTopTXN()

◆ ReorderBufferLargestTXN()

◆ ReorderBufferPrepare()

◆ ReorderBufferProcessPartialChange()

◆ ReorderBufferProcessTXN()

◆ ReorderBufferProcessXid()

◆ ReorderBufferQueueChange()

◆ ReorderBufferQueueMessage()

◆ ReorderBufferRememberPrepareInfo()

◆ ReorderBufferReplay()

◆ ReorderBufferResetTXN()

◆ ReorderBufferRestoreChange()

◆ ReorderBufferRestoreChanges()

◆ ReorderBufferRestoreCleanup()

◆ ReorderBufferReturnChange()