reorderbuffer.c File Reference

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

Include dependency graph for reorderbuffer.c:

Go to the source code of this file.

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

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

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

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

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

Macro Definition Documentation

CHANGES_THRESHOLD

#define CHANGES_THRESHOLD   100

IsInsertOrUpdate

#define IsInsertOrUpdate

(

action

)

Value:

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

Definition at line 193 of file reorderbuffer.c.

IsSpecConfirmOrAbort

#define IsSpecConfirmOrAbort

(

action

)

Value:

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

Definition at line 188 of file reorderbuffer.c.

IsSpecInsert

#define IsSpecInsert

(

action

)

Value:

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

Definition at line 184 of file reorderbuffer.c.

Typedef Documentation

ReorderBufferDiskChange

typedef struct ReorderBufferDiskChange ReorderBufferDiskChange

ReorderBufferIterTXNEntry

typedef struct ReorderBufferIterTXNEntry ReorderBufferIterTXNEntry

ReorderBufferIterTXNState

typedef struct ReorderBufferIterTXNState ReorderBufferIterTXNState

ReorderBufferToastEnt

typedef struct ReorderBufferToastEnt ReorderBufferToastEnt

ReorderBufferTupleCidEnt

typedef struct ReorderBufferTupleCidEnt ReorderBufferTupleCidEnt

ReorderBufferTupleCidKey

typedef struct ReorderBufferTupleCidKey ReorderBufferTupleCidKey

ReorderBufferTXNByIdEnt

typedef struct ReorderBufferTXNByIdEnt ReorderBufferTXNByIdEnt

RewriteMappingFile

typedef struct RewriteMappingFile RewriteMappingFile

TXNEntryFile

typedef struct TXNEntryFile TXNEntryFile

Function Documentation

ApplyLogicalMappingFile()

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

static

Definition at line 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, ReorderBufferTXN::invalidations, MemoryContextSwitchTo(), ReorderBufferTXN::ninvalidations, palloc(), rbtxn_get_toptxn, REORDER_BUFFER_CHANGE_INVALIDATION, ReorderBufferGetChange(), ReorderBufferQueueChange(), ReorderBufferTXNByXid(), and repalloc().

Referenced by xact_decode().

ReorderBufferAddNewCommandId()

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

Definition at line 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::data, dlist_push_tail(), ReorderBufferChange::lsn, ReorderBufferChange::node, ReorderBufferTXN::ntuplecids, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferGetChange(), ReorderBufferTXNByXid(), ReorderBufferChange::tuplecid, ReorderBufferTXN::tuplecids, and ReorderBufferChange::txn.

Referenced by SnapBuildProcessNewCid().

ReorderBufferAddSnapshot()

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

Definition at line 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, ReorderBuffer::message, ReorderBufferChange::msg, and ReorderBuffer::stream_message.

Referenced by ReorderBufferProcessTXN().

ReorderBufferApplyTruncate()

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

inlinestatic

Definition at line 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;
        }
    }
}

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

Referenced by ReorderBufferProcessTXN().

ReorderBufferCanStartStreaming()

static bool ReorderBufferCanStartStreaming ( ReorderBuffer *  rb )

inlinestatic

Definition at line 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;
}

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

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

ReorderBufferCheckMemoryLimit()

static void ReorderBufferCheckMemoryLimit ( ReorderBuffer *  rb )

static

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

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

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

ReorderBufferCommit()

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

Definition at line 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;
 
    /*
     * subxcnt isn't decreased when subtransactions abort, so count manually.
     * Since it's an upper boundary it is safe to use it for the allocation
     * above.
     */
    snap->subxcnt = 1;
 
    dlist_foreach(iter, &txn->subtxns)
    {
        ReorderBufferTXN *sub_txn;
 
        sub_txn = dlist_container(ReorderBufferTXN, node, iter.cur);
        snap->subxip[i++] = sub_txn->xid;
        snap->subxcnt++;
    }
 
    /* sort so we can bsearch() later */
    qsort(snap->subxip, snap->subxcnt, sizeof(TransactionId), xidComparator);
 
    /* store the specified current CommandId */
    snap->curcid = cid;
 
    return snap;
}

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

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

ReorderBufferExecuteInvalidations()

static void ReorderBufferExecuteInvalidations ( uint32  nmsgs, SharedInvalidationMessage *  msgs  )

static

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

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

Referenced by DecodeAbort(), and DecodeCommit().

ReorderBufferForget()

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

Definition at line 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 context, ReorderBuffer::context, ReplicationSlot::data, MemoryContextDelete(), MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, and ReorderBufferCleanupSerializedTXNs().

Referenced by FreeDecodingContext().

ReorderBufferFreeSnap()

static void ReorderBufferFreeSnap ( ReorderBuffer *  rb, Snapshot  snap  )

static

Definition at line 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::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();
}

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

Referenced by ReorderBufferReplay(), and ReorderBufferStreamTXN().

ReorderBufferProcessXid()

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

Definition at line 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(), ReorderBuffer::message, ReorderBufferChange::msg, palloc(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, pstrdup(), REORDER_BUFFER_CHANGE_MESSAGE, ReorderBufferGetChange(), ReorderBufferQueueChange(), ReorderBufferTXNByXid(), SetupHistoricSnapshot(), and TeardownHistoricSnapshot().

Referenced by logicalmsg_decode().

ReorderBufferRememberPrepareInfo()

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

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

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

Referenced by ReorderBufferRestoreChanges().

ReorderBufferRestoreChanges()

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

static

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

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

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

ReorderBufferReturnRelids()

void ReorderBufferReturnRelids	(	ReorderBuffer *	rb,
		Oid *	relids
	)

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

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

Referenced by ReorderBufferSerializeTXN().

ReorderBufferSerializedPath()

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

static

Definition at line 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(), size, ReorderBufferTXN::size, ReorderBuffer::spillBytes, ReorderBuffer::spillCount, ReorderBuffer::spillTxns, ReorderBufferTXN::subtxns, ReorderBufferTXN::txn_flags, UpdateDecodingStats(), wal_segment_size, ReorderBufferTXN::xid, XLByteInSeg, and XLByteToSeg.

Referenced by ReorderBufferCheckMemoryLimit(), and ReorderBufferIterTXNInit().

ReorderBufferSetBaseSnapshot()

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

Definition at line 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, ReorderBuffer::streamBytes, ReorderBuffer::streamCount, ReorderBuffer::streamTxns, ReorderBufferTXN::subtxns, ReorderBufferTXN::total_size, and UpdateDecodingStats().

Referenced by ReorderBufferCheckMemoryLimit(), ReorderBufferProcessPartialChange(), and ReorderBufferStreamCommit().

ReorderBufferToastAppendChunk()

static void ReorderBufferToastAppendChunk ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, Relation  relation, ReorderBufferChange *  change  )

static

Definition at line 4709 of file reorderbuffer.c.

{
    ReorderBufferToastEnt *ent;
    HeapTuple   newtup;
    bool        found;
    int32       chunksize;
    bool        isnull;
    Pointer     chunk;
    TupleDesc   desc = RelationGetDescr(relation);
    Oid         chunk_id;
    int32       chunk_seq;
 
    if (txn->toast_hash == NULL)
        ReorderBufferToastInitHash(rb, txn);
 
    Assert(IsToastRelation(relation));
 
    newtup = change->data.tp.newtuple;
    chunk_id = DatumGetObjectId(fastgetattr(newtup, 1, desc, &isnull));
    Assert(!isnull);
    chunk_seq = DatumGetInt32(fastgetattr(newtup, 2, desc, &isnull));
    Assert(!isnull);
 
    ent = (ReorderBufferToastEnt *)
        hash_search(txn->toast_hash, &chunk_id, HASH_ENTER, &found);
 
    if (!found)
    {
        Assert(ent->chunk_id == chunk_id);
        ent->num_chunks = 0;
        ent->last_chunk_seq = 0;
        ent->size = 0;
        ent->reconstructed = NULL;
        dlist_init(&ent->chunks);
 
        if (chunk_seq != 0)
            elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq 0",
                 chunk_seq, chunk_id);
    }
    else if (found && chunk_seq != ent->last_chunk_seq + 1)
        elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq %d",
             chunk_seq, chunk_id, ent->last_chunk_seq + 1);
 
    chunk = DatumGetPointer(fastgetattr(newtup, 3, desc, &isnull));
    Assert(!isnull);
 
    /* calculate size so we can allocate the right size at once later */
    if (!VARATT_IS_EXTENDED(chunk))
        chunksize = VARSIZE(chunk) - VARHDRSZ;
    else if (VARATT_IS_SHORT(chunk))
        /* could happen due to heap_form_tuple doing its thing */
        chunksize = VARSIZE_SHORT(chunk) - VARHDRSZ_SHORT;
    else
        elog(ERROR, "unexpected type of toast chunk");
 
    ent->size += chunksize;
    ent->last_chunk_seq = chunk_seq;
    ent->num_chunks++;
    dlist_push_tail(&ent->chunks, &change->node);
}

References Assert, chunk, ReorderBufferToastEnt::chunk_id, ReorderBufferToastEnt::chunks, ReorderBufferChange::data, DatumGetInt32(), DatumGetObjectId(), DatumGetPointer(), dlist_init(), dlist_push_tail(), elog, ERROR, fastgetattr(), HASH_ENTER, hash_search(), IsToastRelation(), ReorderBufferToastEnt::last_chunk_seq, ReorderBufferChange::node, ReorderBufferToastEnt::num_chunks, ReorderBufferToastEnt::reconstructed, RelationGetDescr, ReorderBufferToastInitHash(), ReorderBufferToastEnt::size, ReorderBufferTXN::toast_hash, ReorderBufferChange::tp, VARATT_IS_EXTENDED, VARATT_IS_SHORT, VARHDRSZ, VARHDRSZ_SHORT, VARSIZE, and VARSIZE_SHORT.

Referenced by ReorderBufferProcessTXN().

ReorderBufferToastInitHash()

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

static

Definition at line 4689 of file reorderbuffer.c.

{
    HASHCTL     hash_ctl;
 
    Assert(txn->toast_hash == NULL);
 
    hash_ctl.keysize = sizeof(Oid);
    hash_ctl.entrysize = sizeof(ReorderBufferToastEnt);
    hash_ctl.hcxt = rb->context;
    txn->toast_hash = hash_create("ReorderBufferToastHash", 5, &hash_ctl,
                                  HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
}

References Assert, ReorderBuffer::context, HASHCTL::entrysize, HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASHCTL::hcxt, HASHCTL::keysize, and ReorderBufferTXN::toast_hash.

Referenced by ReorderBufferToastAppendChunk().

ReorderBufferToastReplace()

static void ReorderBufferToastReplace ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, Relation  relation, ReorderBufferChange *  change  )

static

Definition at line 4792 of file reorderbuffer.c.

{
    TupleDesc   desc;
    int         natt;
    Datum      *attrs;
    bool       *isnull;
    bool       *free;
    HeapTuple   tmphtup;
    Relation    toast_rel;
    TupleDesc   toast_desc;
    MemoryContext oldcontext;
    HeapTuple   newtup;
    Size        old_size;
 
    /* no toast tuples changed */
    if (txn->toast_hash == NULL)
        return;
 
    /*
     * We're going to modify the size of the change. So, to make sure the
     * accounting is correct we record the current change size and then after
     * re-computing the change we'll subtract the recorded size and then
     * re-add the new change size at the end. We don't immediately subtract
     * the old size because if there is any error before we add the new size,
     * we will release the changes and that will update the accounting info
     * (subtracting the size from the counters). And we don't want to
     * underflow there.
     */
    old_size = ReorderBufferChangeSize(change);
 
    oldcontext = MemoryContextSwitchTo(rb->context);
 
    /* we should only have toast tuples in an INSERT or UPDATE */
    Assert(change->data.tp.newtuple);
 
    desc = RelationGetDescr(relation);
 
    toast_rel = RelationIdGetRelation(relation->rd_rel->reltoastrelid);
    if (!RelationIsValid(toast_rel))
        elog(ERROR, "could not open toast relation with OID %u (base relation \"%s\")",
             relation->rd_rel->reltoastrelid, RelationGetRelationName(relation));
 
    toast_desc = RelationGetDescr(toast_rel);
 
    /* should we allocate from stack instead? */
    attrs = palloc0(sizeof(Datum) * desc->natts);
    isnull = palloc0(sizeof(bool) * desc->natts);
    free = palloc0(sizeof(bool) * desc->natts);
 
    newtup = change->data.tp.newtuple;
 
    heap_deform_tuple(newtup, desc, attrs, isnull);
 
    for (natt = 0; natt < desc->natts; natt++)
    {
        Form_pg_attribute attr = TupleDescAttr(desc, natt);
        ReorderBufferToastEnt *ent;
        struct varlena *varlena;
 
        /* va_rawsize is the size of the original datum -- including header */
        struct varatt_external toast_pointer;
        struct varatt_indirect redirect_pointer;
        struct varlena *new_datum = NULL;
        struct varlena *reconstructed;
        dlist_iter  it;
        Size        data_done = 0;
 
        /* system columns aren't toasted */
        if (attr->attnum < 0)
            continue;
 
        if (attr->attisdropped)
            continue;
 
        /* not a varlena datatype */
        if (attr->attlen != -1)
            continue;
 
        /* no data */
        if (isnull[natt])
            continue;
 
        /* ok, we know we have a toast datum */
        varlena = (struct varlena *) DatumGetPointer(attrs[natt]);
 
        /* no need to do anything if the tuple isn't external */
        if (!VARATT_IS_EXTERNAL(varlena))
            continue;
 
        VARATT_EXTERNAL_GET_POINTER(toast_pointer, varlena);
 
        /*
         * Check whether the toast tuple changed, replace if so.
         */
        ent = (ReorderBufferToastEnt *)
            hash_search(txn->toast_hash,
                        &toast_pointer.va_valueid,
                        HASH_FIND,
                        NULL);
        if (ent == NULL)
            continue;
 
        new_datum =
            (struct varlena *) palloc0(INDIRECT_POINTER_SIZE);
 
        free[natt] = true;
 
        reconstructed = palloc0(toast_pointer.va_rawsize);
 
        ent->reconstructed = reconstructed;
 
        /* stitch toast tuple back together from its parts */
        dlist_foreach(it, &ent->chunks)
        {
            bool        cisnull;
            ReorderBufferChange *cchange;
            HeapTuple   ctup;
            Pointer     chunk;
 
            cchange = dlist_container(ReorderBufferChange, node, it.cur);
            ctup = cchange->data.tp.newtuple;
            chunk = DatumGetPointer(fastgetattr(ctup, 3, toast_desc, &cisnull));
 
            Assert(!cisnull);
            Assert(!VARATT_IS_EXTERNAL(chunk));
            Assert(!VARATT_IS_SHORT(chunk));
 
            memcpy(VARDATA(reconstructed) + data_done,
                   VARDATA(chunk),
                   VARSIZE(chunk) - VARHDRSZ);
            data_done += VARSIZE(chunk) - VARHDRSZ;
        }
        Assert(data_done == VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer));
 
        /* make sure its marked as compressed or not */
        if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
            SET_VARSIZE_COMPRESSED(reconstructed, data_done + VARHDRSZ);
        else
            SET_VARSIZE(reconstructed, data_done + VARHDRSZ);
 
        memset(&redirect_pointer, 0, sizeof(redirect_pointer));
        redirect_pointer.pointer = reconstructed;
 
        SET_VARTAG_EXTERNAL(new_datum, VARTAG_INDIRECT);
        memcpy(VARDATA_EXTERNAL(new_datum), &redirect_pointer,
               sizeof(redirect_pointer));
 
        attrs[natt] = PointerGetDatum(new_datum);
    }
 
    /*
     * Build tuple in separate memory & copy tuple back into the tuplebuf
     * passed to the output plugin. We can't directly heap_fill_tuple() into
     * the tuplebuf because attrs[] will point back into the current content.
     */
    tmphtup = heap_form_tuple(desc, attrs, isnull);
    Assert(newtup->t_len <= MaxHeapTupleSize);
    Assert(newtup->t_data == (HeapTupleHeader) ((char *) newtup + HEAPTUPLESIZE));
 
    memcpy(newtup->t_data, tmphtup->t_data, tmphtup->t_len);
    newtup->t_len = tmphtup->t_len;
 
    /*
     * free resources we won't further need, more persistent stuff will be
     * free'd in ReorderBufferToastReset().
     */
    RelationClose(toast_rel);
    pfree(tmphtup);
    for (natt = 0; natt < desc->natts; natt++)
    {
        if (free[natt])
            pfree(DatumGetPointer(attrs[natt]));
    }
    pfree(attrs);
    pfree(free);
    pfree(isnull);
 
    MemoryContextSwitchTo(oldcontext);
 
    /* subtract the old change size */
    ReorderBufferChangeMemoryUpdate(rb, change, NULL, false, old_size);
    /* now add the change back, with the correct size */
    ReorderBufferChangeMemoryUpdate(rb, change, NULL, true,
                                    ReorderBufferChangeSize(change));
}

References Assert, chunk, ReorderBufferToastEnt::chunks, ReorderBuffer::context, dlist_iter::cur, ReorderBufferChange::data, DatumGetPointer(), dlist_container, dlist_foreach, elog, ERROR, fastgetattr(), free, HASH_FIND, hash_search(), heap_deform_tuple(), heap_form_tuple(), HEAPTUPLESIZE, INDIRECT_POINTER_SIZE, MaxHeapTupleSize, MemoryContextSwitchTo(), TupleDescData::natts, palloc0(), pfree(), varatt_indirect::pointer, PointerGetDatum(), RelationData::rd_rel, ReorderBufferToastEnt::reconstructed, RelationClose(), RelationGetDescr, RelationGetRelationName, RelationIdGetRelation(), RelationIsValid, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), SET_VARSIZE, SET_VARSIZE_COMPRESSED, SET_VARTAG_EXTERNAL, HeapTupleData::t_data, HeapTupleData::t_len, ReorderBufferTXN::toast_hash, ReorderBufferChange::tp, TupleDescAttr, varatt_external::va_rawsize, varatt_external::va_valueid, VARATT_EXTERNAL_GET_EXTSIZE, VARATT_EXTERNAL_GET_POINTER, VARATT_EXTERNAL_IS_COMPRESSED, VARATT_IS_EXTERNAL, VARATT_IS_SHORT, VARDATA, VARDATA_EXTERNAL, VARHDRSZ, VARSIZE, and VARTAG_INDIRECT.

Referenced by ReorderBufferProcessTXN().

ReorderBufferToastReset()

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

static

Definition at line 4983 of file reorderbuffer.c.

{
    HASH_SEQ_STATUS hstat;
    ReorderBufferToastEnt *ent;
 
    if (txn->toast_hash == NULL)
        return;
 
    /* sequentially walk over the hash and free everything */
    hash_seq_init(&hstat, txn->toast_hash);
    while ((ent = (ReorderBufferToastEnt *) hash_seq_search(&hstat)) != NULL)
    {
        dlist_mutable_iter it;
 
        if (ent->reconstructed != NULL)
            pfree(ent->reconstructed);
 
        dlist_foreach_modify(it, &ent->chunks)
        {
            ReorderBufferChange *change =
                dlist_container(ReorderBufferChange, node, it.cur);
 
            dlist_delete(&change->node);
            ReorderBufferReturnChange(rb, change, true);
        }
    }
 
    hash_destroy(txn->toast_hash);
    txn->toast_hash = NULL;
}

References ReorderBufferToastEnt::chunks, dlist_mutable_iter::cur, dlist_container, dlist_delete(), dlist_foreach_modify, hash_destroy(), hash_seq_init(), hash_seq_search(), ReorderBufferChange::node, pfree(), ReorderBufferToastEnt::reconstructed, ReorderBufferReturnChange(), and ReorderBufferTXN::toast_hash.

Referenced by ReorderBufferProcessTXN(), ReorderBufferResetTXN(), and ReorderBufferReturnTXN().

ReorderBufferTransferSnapToParent()

static void ReorderBufferTransferSnapToParent ( ReorderBufferTXN *  txn, ReorderBufferTXN *  subtxn  )

static

Definition at line 1143 of file reorderbuffer.c.

{
    Assert(subtxn->toplevel_xid == txn->xid);
 
    if (subtxn->base_snapshot != NULL)
    {
        if (txn->base_snapshot == NULL ||
            subtxn->base_snapshot_lsn < txn->base_snapshot_lsn)
        {
            /*
             * If the toplevel transaction already has a base snapshot but
             * it's newer than the subxact's, purge it.
             */
            if (txn->base_snapshot != NULL)
            {
                SnapBuildSnapDecRefcount(txn->base_snapshot);
                dlist_delete(&txn->base_snapshot_node);
            }
 
            /*
             * The snapshot is now the top transaction's; transfer it, and
             * adjust the list position of the top transaction in the list by
             * moving it to where the subtransaction is.
             */
            txn->base_snapshot = subtxn->base_snapshot;
            txn->base_snapshot_lsn = subtxn->base_snapshot_lsn;
            dlist_insert_before(&subtxn->base_snapshot_node,
                                &txn->base_snapshot_node);
 
            /*
             * The subtransaction doesn't have a snapshot anymore (so it
             * mustn't be in the list.)
             */
            subtxn->base_snapshot = NULL;
            subtxn->base_snapshot_lsn = InvalidXLogRecPtr;
            dlist_delete(&subtxn->base_snapshot_node);
        }
        else
        {
            /* Base snap of toplevel is fine, so subxact's is not needed */
            SnapBuildSnapDecRefcount(subtxn->base_snapshot);
            dlist_delete(&subtxn->base_snapshot_node);
            subtxn->base_snapshot = NULL;
            subtxn->base_snapshot_lsn = InvalidXLogRecPtr;
        }
    }
}

References Assert, ReorderBufferTXN::base_snapshot, ReorderBufferTXN::base_snapshot_lsn, ReorderBufferTXN::base_snapshot_node, dlist_delete(), dlist_insert_before(), InvalidXLogRecPtr, SnapBuildSnapDecRefcount(), ReorderBufferTXN::toplevel_xid, and ReorderBufferTXN::xid.

Referenced by ReorderBufferAssignChild(), and ReorderBufferStreamTXN().

ReorderBufferTruncateTXN()

static void ReorderBufferTruncateTXN ( ReorderBuffer *  rb, ReorderBufferTXN *  txn, bool  txn_prepared  )

static

Definition at line 1633 of file reorderbuffer.c.

{
    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);
 
        ReorderBufferTruncateTXN(rb, subtxn, txn_prepared);
    }
 
    /* 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);
 
        /* remove the change from it's containing list */
        dlist_delete(&change->node);
 
        /*
         * 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);
 
    /*
     * Mark the transaction as streamed.
     *
     * The top-level transaction, is marked as streamed always, even if it
     * does not contain any changes (that is, when all the changes are in
     * subtransactions).
     *
     * For subtransactions, we only mark them as streamed when there are
     * changes in them.
     *
     * We do it this way because of aborts - we don't want to send aborts for
     * XIDs the downstream is not aware of. And of course, it always knows
     * about the toplevel xact (we send the XID in all messages), but we never
     * stream XIDs of empty subxacts.
     */
    if ((!txn_prepared) && (rbtxn_is_toptxn(txn) || (txn->nentries_mem != 0)))
        txn->txn_flags |= RBTXN_IS_STREAMED;
 
    if (txn_prepared)
    {
        /*
         * If this is a prepared txn, 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);
 
            /* Remove the change from its containing list. */
            dlist_delete(&change->node);
 
            ReorderBufferReturnChange(rb, change, true);
        }
    }
 
    /*
     * Destroy the (relfilelocator, ctid) hashtable, so that we don't leak any
     * memory. We could also keep the hash table and update it with new ctid
     * values, but this seems simpler and good enough for now.
     */
    if (txn->tuplecid_hash != NULL)
    {
        hash_destroy(txn->tuplecid_hash);
        txn->tuplecid_hash = NULL;
    }
 
    /* If this txn is serialized then clean the disk space. */
    if (rbtxn_is_serialized(txn))
    {
        ReorderBufferRestoreCleanup(rb, txn);
        txn->txn_flags &= ~RBTXN_IS_SERIALIZED;
 
        /*
         * We set this flag to indicate if the transaction is ever serialized.
         * We need this to accurately update the stats as otherwise the same
         * transaction can be counted as serialized multiple times.
         */
        txn->txn_flags |= RBTXN_IS_SERIALIZED_CLEAR;
    }
 
    /* also reset the number of entries in the transaction */
    txn->nentries_mem = 0;
    txn->nentries = 0;
}

References ReorderBufferChange::action, Assert, ReorderBufferTXN::changes, dlist_mutable_iter::cur, dlist_container, dlist_delete(), dlist_foreach_modify, hash_destroy(), ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, ReorderBufferChange::node, ReorderBufferTXN::nsubtxns, rbtxn_is_known_subxact, RBTXN_IS_SERIALIZED, rbtxn_is_serialized, RBTXN_IS_SERIALIZED_CLEAR, RBTXN_IS_STREAMED, rbtxn_is_toptxn, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferChangeMemoryUpdate(), ReorderBufferChangeSize(), ReorderBufferRestoreCleanup(), ReorderBufferReturnChange(), ReorderBufferTXN::subtxns, ReorderBufferTXN::tuplecid_hash, ReorderBufferTXN::tuplecids, ReorderBufferChange::txn, and ReorderBufferTXN::txn_flags.

Referenced by ReorderBufferProcessTXN(), ReorderBufferResetTXN(), and ReorderBufferStreamCommit().

ReorderBufferTXNByXid()

static ReorderBufferTXN * ReorderBufferTXNByXid ( ReorderBuffer *  rb, TransactionId  xid, bool  create, bool *  is_new, XLogRecPtr  lsn, bool  create_as_top  )

static

Definition at line 631 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
    ReorderBufferTXNByIdEnt *ent;
    bool        found;
 
    Assert(TransactionIdIsValid(xid));
 
    /*
     * Check the one-entry lookup cache first
     */
    if (TransactionIdIsValid(rb->by_txn_last_xid) &&
        rb->by_txn_last_xid == xid)
    {
        txn = rb->by_txn_last_txn;
 
        if (txn != NULL)
        {
            /* found it, and it's valid */
            if (is_new)
                *is_new = false;
            return txn;
        }
 
        /*
         * cached as non-existent, and asked not to create? Then nothing else
         * to do.
         */
        if (!create)
            return NULL;
        /* otherwise fall through to create it */
    }
 
    /*
     * If the cache wasn't hit or it yielded a "does-not-exist" and we want to
     * create an entry.
     */
 
    /* search the lookup table */
    ent = (ReorderBufferTXNByIdEnt *)
        hash_search(rb->by_txn,
                    &xid,
                    create ? HASH_ENTER : HASH_FIND,
                    &found);
    if (found)
        txn = ent->txn;
    else if (create)
    {
        /* initialize the new entry, if creation was requested */
        Assert(ent != NULL);
        Assert(lsn != InvalidXLogRecPtr);
 
        ent->txn = ReorderBufferGetTXN(rb);
        ent->txn->xid = xid;
        txn = ent->txn;
        txn->first_lsn = lsn;
        txn->restart_decoding_lsn = rb->current_restart_decoding_lsn;
 
        if (create_as_top)
        {
            dlist_push_tail(&rb->toplevel_by_lsn, &txn->node);
            AssertTXNLsnOrder(rb);
        }
    }
    else
        txn = NULL;             /* not found and not asked to create */
 
    /* update cache */
    rb->by_txn_last_xid = xid;
    rb->by_txn_last_txn = txn;
 
    if (is_new)
        *is_new = !found;
 
    Assert(!create || txn != NULL);
    return txn;
}

References Assert, AssertTXNLsnOrder(), ReorderBuffer::by_txn, ReorderBuffer::by_txn_last_txn, ReorderBuffer::by_txn_last_xid, ReorderBuffer::current_restart_decoding_lsn, dlist_push_tail(), ReorderBufferTXN::first_lsn, HASH_ENTER, HASH_FIND, hash_search(), InvalidXLogRecPtr, ReorderBufferTXN::node, ReorderBufferGetTXN(), ReorderBufferTXN::restart_decoding_lsn, ReorderBuffer::toplevel_by_lsn, TransactionIdIsValid, ReorderBufferTXNByIdEnt::txn, and ReorderBufferTXN::xid.

Referenced by ReorderBufferAbort(), ReorderBufferAddInvalidations(), ReorderBufferAddNewTupleCids(), ReorderBufferAssignChild(), ReorderBufferCommit(), ReorderBufferCommitChild(), ReorderBufferFinishPrepared(), ReorderBufferForget(), ReorderBufferInvalidate(), ReorderBufferPrepare(), ReorderBufferProcessXid(), ReorderBufferQueueChange(), ReorderBufferQueueMessage(), ReorderBufferRememberPrepareInfo(), ReorderBufferSetBaseSnapshot(), ReorderBufferSkipPrepare(), ReorderBufferXidHasBaseSnapshot(), ReorderBufferXidHasCatalogChanges(), and ReorderBufferXidSetCatalogChanges().

ReorderBufferTXNSizeCompare()

static int ReorderBufferTXNSizeCompare ( const pairingheap_node *  a, const pairingheap_node *  b, void *  arg  )

static

Definition at line 3537 of file reorderbuffer.c.

{
    const ReorderBufferTXN *ta = pairingheap_const_container(ReorderBufferTXN, txn_node, a);
    const ReorderBufferTXN *tb = pairingheap_const_container(ReorderBufferTXN, txn_node, b);
 
    if (ta->size < tb->size)
        return -1;
    if (ta->size > tb->size)
        return 1;
    return 0;
}

References a, b, pairingheap_const_container, and ReorderBufferTXN::size.

Referenced by ReorderBufferAllocate().

ReorderBufferXidHasBaseSnapshot()

bool ReorderBufferXidHasBaseSnapshot	(	ReorderBuffer *	rb,
		TransactionId	xid
	)

Definition at line 3490 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false,
                                NULL, InvalidXLogRecPtr, false);
 
    /* transaction isn't known yet, ergo no snapshot */
    if (txn == NULL)
        return false;
 
    /* a known subtxn? operate on top-level txn instead */
    if (rbtxn_is_known_subxact(txn))
        txn = ReorderBufferTXNByXid(rb, txn->toplevel_xid, false,
                                    NULL, InvalidXLogRecPtr, false);
 
    return txn->base_snapshot != NULL;
}

References ReorderBufferTXN::base_snapshot, InvalidXLogRecPtr, rbtxn_is_known_subxact, ReorderBufferTXNByXid(), and ReorderBufferTXN::toplevel_xid.

Referenced by SnapBuildCommitTxn(), SnapBuildDistributeNewCatalogSnapshot(), and SnapBuildProcessChange().

ReorderBufferXidHasCatalogChanges()

bool ReorderBufferXidHasCatalogChanges	(	ReorderBuffer *	rb,
		TransactionId	xid
	)

Definition at line 3473 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
    if (txn == NULL)
        return false;
 
    return rbtxn_has_catalog_changes(txn);
}

References InvalidXLogRecPtr, rbtxn_has_catalog_changes, and ReorderBufferTXNByXid().

Referenced by SnapBuildXidHasCatalogChanges().

ReorderBufferXidSetCatalogChanges()

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

Definition at line 3400 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
    if (!rbtxn_has_catalog_changes(txn))
    {
        txn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
        dclist_push_tail(&rb->catchange_txns, &txn->catchange_node);
    }
 
    /*
     * Mark top-level transaction as having catalog changes too if one of its
     * children has so that the ReorderBufferBuildTupleCidHash can
     * conveniently check just top-level transaction and decide whether to
     * build the hash table or not.
     */
    if (rbtxn_is_subtxn(txn))
    {
        ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn);
 
        if (!rbtxn_has_catalog_changes(toptxn))
        {
            toptxn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
            dclist_push_tail(&rb->catchange_txns, &toptxn->catchange_node);
        }
    }
}

References ReorderBufferTXN::catchange_node, ReorderBuffer::catchange_txns, dclist_push_tail(), rbtxn_get_toptxn, RBTXN_HAS_CATALOG_CHANGES, rbtxn_has_catalog_changes, rbtxn_is_subtxn, ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.

Referenced by heap_decode(), SnapBuildProcessNewCid(), and xact_decode().

ResolveCminCmaxDuringDecoding()

bool ResolveCminCmaxDuringDecoding	(	HTAB *	tuplecid_data,
		Snapshot	snapshot,
		HeapTuple	htup,
		Buffer	buffer,
		CommandId *	cmin,
		CommandId *	cmax
	)

Definition at line 5275 of file reorderbuffer.c.

{
    ReorderBufferTupleCidKey key;
    ReorderBufferTupleCidEnt *ent;
    ForkNumber  forkno;
    BlockNumber blockno;
    bool        updated_mapping = false;
 
    /*
     * Return unresolved if tuplecid_data is not valid.  That's because when
     * streaming in-progress transactions we may run into tuples with the CID
     * before actually decoding them.  Think e.g. about INSERT followed by
     * TRUNCATE, where the TRUNCATE may not be decoded yet when applying the
     * INSERT.  So in such cases, we assume the CID is from the future
     * command.
     */
    if (tuplecid_data == NULL)
        return false;
 
    /* be careful about padding */
    memset(&key, 0, sizeof(key));
 
    Assert(!BufferIsLocal(buffer));
 
    /*
     * get relfilelocator from the buffer, no convenient way to access it
     * other than that.
     */
    BufferGetTag(buffer, &key.rlocator, &forkno, &blockno);
 
    /* tuples can only be in the main fork */
    Assert(forkno == MAIN_FORKNUM);
    Assert(blockno == ItemPointerGetBlockNumber(&htup->t_self));
 
    ItemPointerCopy(&htup->t_self,
                    &key.tid);
 
restart:
    ent = (ReorderBufferTupleCidEnt *)
        hash_search(tuplecid_data, &key, HASH_FIND, NULL);
 
    /*
     * failed to find a mapping, check whether the table was rewritten and
     * apply mapping if so, but only do that once - there can be no new
     * mappings while we are in here since we have to hold a lock on the
     * relation.
     */
    if (ent == NULL && !updated_mapping)
    {
        UpdateLogicalMappings(tuplecid_data, htup->t_tableOid, snapshot);
        /* now check but don't update for a mapping again */
        updated_mapping = true;
        goto restart;
    }
    else if (ent == NULL)
        return false;
 
    if (cmin)
        *cmin = ent->cmin;
    if (cmax)
        *cmax = ent->cmax;
    return true;
}

References Assert, BufferGetTag(), BufferIsLocal, ReorderBufferTupleCidEnt::cmax, ReorderBufferTupleCidEnt::cmin, HASH_FIND, hash_search(), ItemPointerCopy(), ItemPointerGetBlockNumber(), sort-test::key, MAIN_FORKNUM, HeapTupleData::t_self, HeapTupleData::t_tableOid, tuplecid_data, and UpdateLogicalMappings().

Referenced by HeapTupleSatisfiesHistoricMVCC().

SetupCheckXidLive()

static void SetupCheckXidLive ( TransactionId  xid )

inlinestatic

Definition at line 1971 of file reorderbuffer.c.

{
    /*
     * If the input transaction id is already set as a CheckXidAlive then
     * nothing to do.
     */
    if (TransactionIdEquals(CheckXidAlive, xid))
        return;
 
    /*
     * setup CheckXidAlive if it's not committed yet.  We don't check if the
     * xid is aborted.  That will happen during catalog access.
     */
    if (!TransactionIdDidCommit(xid))
        CheckXidAlive = xid;
    else
        CheckXidAlive = InvalidTransactionId;
}

References CheckXidAlive, InvalidTransactionId, TransactionIdDidCommit(), and TransactionIdEquals.

Referenced by ReorderBufferProcessTXN().

StartupReorderBuffer()

void StartupReorderBuffer

(

void

)

Definition at line 4655 of file reorderbuffer.c.

{
    DIR        *logical_dir;
    struct dirent *logical_de;
 
    logical_dir = AllocateDir(PG_REPLSLOT_DIR);
    while ((logical_de = ReadDir(logical_dir, PG_REPLSLOT_DIR)) != NULL)
    {
        if (strcmp(logical_de->d_name, ".") == 0 ||
            strcmp(logical_de->d_name, "..") == 0)
            continue;
 
        /* if it cannot be a slot, skip the directory */
        if (!ReplicationSlotValidateName(logical_de->d_name, DEBUG2))
            continue;
 
        /*
         * ok, has to be a surviving logical slot, iterate and delete
         * everything starting with xid-*
         */
        ReorderBufferCleanupSerializedTXNs(logical_de->d_name);
    }
    FreeDir(logical_dir);
}

References AllocateDir(), dirent::d_name, DEBUG2, FreeDir(), PG_REPLSLOT_DIR, ReadDir(), ReorderBufferCleanupSerializedTXNs(), and ReplicationSlotValidateName().

Referenced by StartupXLOG().

TransactionIdInArray()

static bool TransactionIdInArray ( TransactionId  xid, TransactionId *  xip, Size  num  )

static

Definition at line 5174 of file reorderbuffer.c.

{
    return bsearch(&xid, xip, num,
                   sizeof(TransactionId), xidComparator) != NULL;
}

References xidComparator().

Referenced by UpdateLogicalMappings().

UpdateLogicalMappings()

static void UpdateLogicalMappings ( HTAB *  tuplecid_data, Oid  relid, Snapshot  snapshot  )

static

Definition at line 5197 of file reorderbuffer.c.

{
    DIR        *mapping_dir;
    struct dirent *mapping_de;
    List       *files = NIL;
    ListCell   *file;
    Oid         dboid = IsSharedRelation(relid) ? InvalidOid : MyDatabaseId;
 
    mapping_dir = AllocateDir(PG_LOGICAL_MAPPINGS_DIR);
    while ((mapping_de = ReadDir(mapping_dir, PG_LOGICAL_MAPPINGS_DIR)) != NULL)
    {
        Oid         f_dboid;
        Oid         f_relid;
        TransactionId f_mapped_xid;
        TransactionId f_create_xid;
        XLogRecPtr  f_lsn;
        uint32      f_hi,
                    f_lo;
        RewriteMappingFile *f;
 
        if (strcmp(mapping_de->d_name, ".") == 0 ||
            strcmp(mapping_de->d_name, "..") == 0)
            continue;
 
        /* Ignore files that aren't ours */
        if (strncmp(mapping_de->d_name, "map-", 4) != 0)
            continue;
 
        if (sscanf(mapping_de->d_name, LOGICAL_REWRITE_FORMAT,
                   &f_dboid, &f_relid, &f_hi, &f_lo,
                   &f_mapped_xid, &f_create_xid) != 6)
            elog(ERROR, "could not parse filename \"%s\"", mapping_de->d_name);
 
        f_lsn = ((uint64) f_hi) << 32 | f_lo;
 
        /* mapping for another database */
        if (f_dboid != dboid)
            continue;
 
        /* mapping for another relation */
        if (f_relid != relid)
            continue;
 
        /* did the creating transaction abort? */
        if (!TransactionIdDidCommit(f_create_xid))
            continue;
 
        /* not for our transaction */
        if (!TransactionIdInArray(f_mapped_xid, snapshot->subxip, snapshot->subxcnt))
            continue;
 
        /* ok, relevant, queue for apply */
        f = palloc(sizeof(RewriteMappingFile));
        f->lsn = f_lsn;
        strcpy(f->fname, mapping_de->d_name);
        files = lappend(files, f);
    }
    FreeDir(mapping_dir);
 
    /* sort files so we apply them in LSN order */
    list_sort(files, file_sort_by_lsn);
 
    foreach(file, files)
    {
        RewriteMappingFile *f = (RewriteMappingFile *) lfirst(file);
 
        elog(DEBUG1, "applying mapping: \"%s\" in %u", f->fname,
             snapshot->subxip[0]);
        ApplyLogicalMappingFile(tuplecid_data, relid, f->fname);
        pfree(f);
    }
}

References AllocateDir(), ApplyLogicalMappingFile(), dirent::d_name, DEBUG1, elog, ERROR, file_sort_by_lsn(), RewriteMappingFile::fname, FreeDir(), InvalidOid, IsSharedRelation(), lappend(), lfirst, list_sort(), LOGICAL_REWRITE_FORMAT, RewriteMappingFile::lsn, MyDatabaseId, NIL, palloc(), pfree(), PG_LOGICAL_MAPPINGS_DIR, ReadDir(), SnapshotData::subxcnt, SnapshotData::subxip, TransactionIdDidCommit(), TransactionIdInArray(), and tuplecid_data.

Referenced by ResolveCminCmaxDuringDecoding().

Variable Documentation

debug_logical_replication_streaming

int debug_logical_replication_streaming = DEBUG_LOGICAL_REP_STREAMING_BUFFERED

Definition at line 216 of file reorderbuffer.c.

Referenced by pa_send_data(), and ReorderBufferCheckMemoryLimit().

logical_decoding_work_mem

int logical_decoding_work_mem

Definition at line 212 of file reorderbuffer.c.

Referenced by ReorderBufferCheckMemoryLimit().

max_changes_in_memory

const Size max_changes_in_memory = 4096

static

Definition at line 213 of file reorderbuffer.c.

Referenced by ReorderBufferRestoreChanges().