#include "access/htup_details.h"
#include "lib/ilist.h"
#include "lib/pairingheap.h"
#include "storage/sinval.h"
#include "utils/hsearch.h"
#include "utils/relcache.h"
#include "utils/snapshot.h"
#include "utils/timestamp.h"

Include dependency graph for reorderbuffer.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	ReorderBufferChange

struct	ReorderBufferTXN

struct	ReorderBuffer

Macros
#define	PG_LOGICAL_DIR "pg_logical"

#define	PG_LOGICAL_MAPPINGS_DIR PG_LOGICAL_DIR "/mappings"

#define	PG_LOGICAL_SNAPSHOTS_DIR PG_LOGICAL_DIR "/snapshots"

#define	RBTXN_HAS_CATALOG_CHANGES 0x0001

#define	RBTXN_IS_SUBXACT 0x0002

#define	RBTXN_IS_SERIALIZED 0x0004

#define	RBTXN_IS_SERIALIZED_CLEAR 0x0008

#define	RBTXN_IS_STREAMED 0x0010

#define	RBTXN_HAS_PARTIAL_CHANGE 0x0020

#define	RBTXN_IS_PREPARED 0x0040

#define	RBTXN_SKIPPED_PREPARE 0x0080

#define	RBTXN_HAS_STREAMABLE_CHANGE 0x0100

#define	RBTXN_SENT_PREPARE 0x0200

#define	RBTXN_IS_COMMITTED 0x0400

#define	RBTXN_IS_ABORTED 0x0800

#define	RBTXN_DISTR_INVAL_OVERFLOWED 0x1000

#define	RBTXN_PREPARE_STATUS_MASK (RBTXN_IS_PREPARED \| RBTXN_SKIPPED_PREPARE \| RBTXN_SENT_PREPARE)

#define	rbtxn_has_catalog_changes(txn)

#define	rbtxn_is_known_subxact(txn)

#define	rbtxn_is_serialized(txn)

#define	rbtxn_is_serialized_clear(txn)

#define	rbtxn_has_partial_change(txn)

#define	rbtxn_has_streamable_change(txn)

#define	rbtxn_is_streamed(txn)

#define	rbtxn_is_prepared(txn)

#define	rbtxn_sent_prepare(txn)

#define	rbtxn_is_committed(txn)

#define	rbtxn_is_aborted(txn)

#define	rbtxn_skip_prepared(txn)

#define	rbtxn_distr_inval_overflowed(txn)

#define	rbtxn_is_toptxn(txn)

#define	rbtxn_is_subtxn(txn)

#define	rbtxn_get_toptxn(txn)

Typedefs
typedef enum ReorderBufferChangeType	ReorderBufferChangeType

typedef struct ReorderBufferChange	ReorderBufferChange

typedef struct ReorderBufferTXN	ReorderBufferTXN

typedef struct ReorderBuffer	ReorderBuffer

typedef void(*	ReorderBufferApplyChangeCB) (ReorderBuffer rb, ReorderBufferTXN txn, Relation relation, ReorderBufferChange *change)

typedef void(*	ReorderBufferApplyTruncateCB) (ReorderBuffer rb, ReorderBufferTXN txn, int nrelations, Relation relations[], ReorderBufferChange *change)

typedef void(*	ReorderBufferBeginCB) (ReorderBuffer rb, ReorderBufferTXN txn)

typedef void(*	ReorderBufferCommitCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr commit_lsn)

typedef void(*	ReorderBufferMessageCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr message_lsn, bool transactional, const char prefix, Size sz, const char message)

typedef void(*	ReorderBufferBeginPrepareCB) (ReorderBuffer rb, ReorderBufferTXN txn)

typedef void(*	ReorderBufferPrepareCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr prepare_lsn)

typedef void(*	ReorderBufferCommitPreparedCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr commit_lsn)

typedef void(*	ReorderBufferRollbackPreparedCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)

typedef void(*	ReorderBufferStreamStartCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr first_lsn)

typedef void(*	ReorderBufferStreamStopCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr last_lsn)

typedef void(*	ReorderBufferStreamAbortCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr abort_lsn)

typedef void(*	ReorderBufferStreamPrepareCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr prepare_lsn)

typedef void(*	ReorderBufferStreamCommitCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr commit_lsn)

typedef void(*	ReorderBufferStreamChangeCB) (ReorderBuffer rb, ReorderBufferTXN txn, Relation relation, ReorderBufferChange *change)

typedef void(*	ReorderBufferStreamMessageCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr message_lsn, bool transactional, const char prefix, Size sz, const char message)

typedef void(*	ReorderBufferStreamTruncateCB) (ReorderBuffer rb, ReorderBufferTXN txn, int nrelations, Relation relations[], ReorderBufferChange *change)

typedef void(*	ReorderBufferUpdateProgressTxnCB) (ReorderBuffer rb, ReorderBufferTXN txn, XLogRecPtr lsn)

Enumerations
enum	DebugLogicalRepStreamingMode { DEBUG_LOGICAL_REP_STREAMING_BUFFERED , DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE }

enum	ReorderBufferChangeType { REORDER_BUFFER_CHANGE_INSERT , REORDER_BUFFER_CHANGE_UPDATE , REORDER_BUFFER_CHANGE_DELETE , REORDER_BUFFER_CHANGE_MESSAGE , REORDER_BUFFER_CHANGE_INVALIDATION , REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT , REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID , REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID , REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT , REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM , REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT , REORDER_BUFFER_CHANGE_TRUNCATE }

Functions
ReorderBuffer *	ReorderBufferAllocate (void)

void	ReorderBufferFree (ReorderBuffer *rb)

HeapTuple	ReorderBufferAllocTupleBuf (ReorderBuffer *rb, Size tuple_len)

void	ReorderBufferFreeTupleBuf (HeapTuple tuple)

ReorderBufferChange *	ReorderBufferAllocChange (ReorderBuffer *rb)

void	ReorderBufferFreeChange (ReorderBuffer rb, ReorderBufferChange change, bool upd_mem)

Oid *	ReorderBufferAllocRelids (ReorderBuffer *rb, int nrelids)

void	ReorderBufferFreeRelids (ReorderBuffer rb, Oid relids)

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)

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

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

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

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

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	ReorderBufferSetBaseSnapshot (ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)

void	ReorderBufferAddSnapshot (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	ReorderBufferAddDistributedInvalidations (ReorderBuffer rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage msgs)

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

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

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

bool	ReorderBufferXidHasCatalogChanges (ReorderBuffer *rb, TransactionId xid)

bool	ReorderBufferXidHasBaseSnapshot (ReorderBuffer *rb, TransactionId xid)

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

void	ReorderBufferSkipPrepare (ReorderBuffer *rb, TransactionId xid)

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

ReorderBufferTXN *	ReorderBufferGetOldestTXN (ReorderBuffer *rb)

TransactionId	ReorderBufferGetOldestXmin (ReorderBuffer *rb)

TransactionId *	ReorderBufferGetCatalogChangesXacts (ReorderBuffer *rb)

void	ReorderBufferSetRestartPoint (ReorderBuffer *rb, XLogRecPtr ptr)

uint32	ReorderBufferGetInvalidations (ReorderBuffer rb, TransactionId xid, SharedInvalidationMessage *msgs)

void	StartupReorderBuffer (void)

Variables
PGDLLIMPORT int	logical_decoding_work_mem

PGDLLIMPORT int	debug_logical_replication_streaming

Macro Definition Documentation

◆ PG_LOGICAL_DIR

#define PG_LOGICAL_DIR "pg_logical"

Definition at line 22 of file reorderbuffer.h.

◆ PG_LOGICAL_MAPPINGS_DIR

#define PG_LOGICAL_MAPPINGS_DIR PG_LOGICAL_DIR "/mappings"

Definition at line 23 of file reorderbuffer.h.

◆ PG_LOGICAL_SNAPSHOTS_DIR

#define PG_LOGICAL_SNAPSHOTS_DIR PG_LOGICAL_DIR "/snapshots"

Definition at line 24 of file reorderbuffer.h.

◆ RBTXN_DISTR_INVAL_OVERFLOWED

#define RBTXN_DISTR_INVAL_OVERFLOWED 0x1000

Definition at line 179 of file reorderbuffer.h.

◆ rbtxn_distr_inval_overflowed

#define rbtxn_distr_inval_overflowed ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_DISTR_INVAL_OVERFLOWED) != 0 \
)

Definition at line 270 of file reorderbuffer.h.

                                         : (txn) \
)
 
typedef struct ReorderBufferTXN
{
    /* See above */
    bits32      txn_flags;
 
    /* The transaction's transaction id, can be a toplevel or sub xid. */
    TransactionId xid;
 
    /* Xid of top-level transaction, if known */
    TransactionId toplevel_xid;
 
    /*
     * Global transaction id required for identification of prepared
     * transactions.
     */
    char       *gid;
 
    /*
     * LSN of the first data carrying, WAL record with knowledge about this
     * xid. This is allowed to *not* be first record adorned with this xid, if
     * the previous records aren't relevant for logical decoding.
     */
    XLogRecPtr  first_lsn;
 
    /* ----
     * LSN of the record that lead to this xact to be prepared or committed or
     * aborted. This can be a
     * * plain commit record
     * * plain commit record, of a parent transaction
     * * prepared transaction
     * * prepared transaction commit
     * * plain abort record
     * * prepared transaction abort
     *
     * This can also become set to earlier values than transaction end when
     * a transaction is spilled to disk; specifically it's set to the LSN of
     * the latest change written to disk so far.
     * ----
     */
    XLogRecPtr  final_lsn;
 
    /*
     * LSN pointing to the end of the commit record + 1.
     */
    XLogRecPtr  end_lsn;
 
    /* Toplevel transaction for this subxact (NULL for top-level). */
    struct ReorderBufferTXN *toptxn;
 
    /*
     * LSN of the last lsn at which snapshot information reside, so we can
     * restart decoding from there and fully recover this transaction from
     * WAL.
     */
    XLogRecPtr  restart_decoding_lsn;
 
    /* origin of the change that caused this transaction */
    ReplOriginId origin_id;
    XLogRecPtr  origin_lsn;
 
    /*
     * Commit or Prepare time, only known when we read the actual commit or
     * prepare record.
     */
    union
    {
        TimestampTz commit_time;
        TimestampTz prepare_time;
        TimestampTz abort_time;
    };
 
    /*
     * The base snapshot is used to decode all changes until either this
     * transaction modifies the catalog, or another catalog-modifying
     * transaction commits.
     */
    Snapshot    base_snapshot;
    XLogRecPtr  base_snapshot_lsn;
    dlist_node  base_snapshot_node; /* link in txns_by_base_snapshot_lsn */
 
    /*
     * Snapshot/CID from the previous streaming run. Only valid for already
     * streamed transactions (NULL/InvalidCommandId otherwise).
     */
    Snapshot    snapshot_now;
    CommandId   command_id;
 
    /*
     * How many ReorderBufferChange's do we have in this txn.
     *
     * Changes in subtransactions are *not* included but tracked separately.
     */
    uint64      nentries;
 
    /*
     * How many of the above entries are stored in memory in contrast to being
     * spilled to disk.
     */
    uint64      nentries_mem;
 
    /*
     * List of ReorderBufferChange structs, including new Snapshots, new
     * CommandIds and command invalidation messages.
     */
    dlist_head  changes;
 
    /*
     * List of (relation, ctid) => (cmin, cmax) mappings for catalog tuples.
     * Those are always assigned to the toplevel transaction. (Keep track of
     * #entries to create a hash of the right size)
     */
    dlist_head  tuplecids;
    uint64      ntuplecids;
 
    /*
     * On-demand built hash for looking up the above values.
     */
    HTAB       *tuplecid_hash;
 
    /*
     * Hash containing (potentially partial) toast entries. NULL if no toast
     * tuples have been found for the current change.
     */
    HTAB       *toast_hash;
 
    /*
     * non-hierarchical list of subtransactions that are *not* aborted. Only
     * used in toplevel transactions.
     */
    dlist_head  subtxns;
    uint32      nsubtxns;
 
    /*
     * Stored cache invalidations. This is not a linked list because we get
     * all the invalidations at once.
     */
    uint32      ninvalidations;
    SharedInvalidationMessage *invalidations;
 
    /*
     * Stores cache invalidation messages distributed by other transactions.
     */
    uint32      ninvalidations_distributed;
    SharedInvalidationMessage *invalidations_distributed;
 
    /* ---
     * Position in one of two lists:
     * * list of subtransactions if we are *known* to be subxact
     * * list of toplevel xacts (can be an as-yet unknown subxact)
     * ---
     */
    dlist_node  node;
 
    /*
     * A node in the list of catalog modifying transactions
     */
    dlist_node  catchange_node;
 
    /*
     * A node in txn_heap
     */
    pairingheap_node txn_node;
 
    /*
     * Size of this transaction (changes currently in memory, in bytes).
     */
    Size        size;
 
    /* Size of top-transaction including sub-transactions. */
    Size        total_size;
 
    /*
     * Private data pointer of the output plugin.
     */
    void       *output_plugin_private;
} ReorderBufferTXN;
 
/* so we can define the callbacks used inside struct ReorderBuffer itself */
typedef struct ReorderBuffer ReorderBuffer;
 
/* change callback signature */
typedef void (*ReorderBufferApplyChangeCB) (ReorderBuffer *rb,
                                            ReorderBufferTXN *txn,
                                            Relation relation,
                                            ReorderBufferChange *change);
 
/* truncate callback signature */
typedef void (*ReorderBufferApplyTruncateCB) (ReorderBuffer *rb,
                                              ReorderBufferTXN *txn,
                                              int nrelations,
                                              Relation relations[],
                                              ReorderBufferChange *change);
 
/* begin callback signature */
typedef void (*ReorderBufferBeginCB) (ReorderBuffer *rb,
                                      ReorderBufferTXN *txn);
 
/* commit callback signature */
typedef void (*ReorderBufferCommitCB) (ReorderBuffer *rb,
                                       ReorderBufferTXN *txn,
                                       XLogRecPtr commit_lsn);
 
/* message callback signature */
typedef void (*ReorderBufferMessageCB) (ReorderBuffer *rb,
                                        ReorderBufferTXN *txn,
                                        XLogRecPtr message_lsn,
                                        bool transactional,
                                        const char *prefix, Size sz,
                                        const char *message);
 
/* begin prepare callback signature */
typedef void (*ReorderBufferBeginPrepareCB) (ReorderBuffer *rb,
                                             ReorderBufferTXN *txn);
 
/* prepare callback signature */
typedef void (*ReorderBufferPrepareCB) (ReorderBuffer *rb,
                                        ReorderBufferTXN *txn,
                                        XLogRecPtr prepare_lsn);
 
/* commit prepared callback signature */
typedef void (*ReorderBufferCommitPreparedCB) (ReorderBuffer *rb,
                                               ReorderBufferTXN *txn,
                                               XLogRecPtr commit_lsn);
 
/* rollback  prepared callback signature */
typedef void (*ReorderBufferRollbackPreparedCB) (ReorderBuffer *rb,
                                                 ReorderBufferTXN *txn,
                                                 XLogRecPtr prepare_end_lsn,
                                                 TimestampTz prepare_time);
 
/* start streaming transaction callback signature */
typedef void (*ReorderBufferStreamStartCB) (ReorderBuffer *rb,
                                            ReorderBufferTXN *txn,
                                            XLogRecPtr first_lsn);
 
/* stop streaming transaction callback signature */
typedef void (*ReorderBufferStreamStopCB) (ReorderBuffer *rb,
                                           ReorderBufferTXN *txn,
                                           XLogRecPtr last_lsn);
 
/* discard streamed transaction callback signature */
typedef void (*ReorderBufferStreamAbortCB) (ReorderBuffer *rb,
                                            ReorderBufferTXN *txn,
                                            XLogRecPtr abort_lsn);
 
/* prepare streamed transaction callback signature */
typedef void (*ReorderBufferStreamPrepareCB) (ReorderBuffer *rb,
                                              ReorderBufferTXN *txn,
                                              XLogRecPtr prepare_lsn);
 
/* commit streamed transaction callback signature */
typedef void (*ReorderBufferStreamCommitCB) (ReorderBuffer *rb,
                                             ReorderBufferTXN *txn,
                                             XLogRecPtr commit_lsn);
 
/* stream change callback signature */
typedef void (*ReorderBufferStreamChangeCB) (ReorderBuffer *rb,
                                             ReorderBufferTXN *txn,
                                             Relation relation,
                                             ReorderBufferChange *change);
 
/* stream message callback signature */
typedef void (*ReorderBufferStreamMessageCB) (ReorderBuffer *rb,
                                              ReorderBufferTXN *txn,
                                              XLogRecPtr message_lsn,
                                              bool transactional,
                                              const char *prefix, Size sz,
                                              const char *message);
 
/* stream truncate callback signature */
typedef void (*ReorderBufferStreamTruncateCB) (ReorderBuffer *rb,
                                               ReorderBufferTXN *txn,
                                               int nrelations,
                                               Relation relations[],
                                               ReorderBufferChange *change);
 
/* update progress txn callback signature */
typedef void (*ReorderBufferUpdateProgressTxnCB) (ReorderBuffer *rb,
                                                  ReorderBufferTXN *txn,
                                                  XLogRecPtr lsn);
 
struct ReorderBuffer
{
    /*
     * xid => ReorderBufferTXN lookup table
     */
    HTAB       *by_txn;
 
    /*
     * Transactions that could be a toplevel xact, ordered by LSN of the first
     * record bearing that xid.
     */
    dlist_head  toplevel_by_lsn;
 
    /*
     * Transactions and subtransactions that have a base snapshot, ordered by
     * LSN of the record which caused us to first obtain the base snapshot.
     * This is not the same as toplevel_by_lsn, because we only set the base
     * snapshot on the first logical-decoding-relevant record (eg. heap
     * writes), whereas the initial LSN could be set by other operations.
     */
    dlist_head  txns_by_base_snapshot_lsn;
 
    /*
     * Transactions and subtransactions that have modified system catalogs.
     */
    dclist_head catchange_txns;
 
    /*
     * one-entry sized cache for by_txn. Very frequently the same txn gets
     * looked up over and over again.
     */
    TransactionId by_txn_last_xid;
    ReorderBufferTXN *by_txn_last_txn;
 
    /*
     * Callbacks to be called when a transactions commits.
     */
    ReorderBufferBeginCB begin;
    ReorderBufferApplyChangeCB apply_change;
    ReorderBufferApplyTruncateCB apply_truncate;
    ReorderBufferCommitCB commit;
    ReorderBufferMessageCB message;
 
    /*
     * Callbacks to be called when streaming a transaction at prepare time.
     */
    ReorderBufferBeginCB begin_prepare;
    ReorderBufferPrepareCB prepare;
    ReorderBufferCommitPreparedCB commit_prepared;
    ReorderBufferRollbackPreparedCB rollback_prepared;
 
    /*
     * Callbacks to be called when streaming a transaction.
     */
    ReorderBufferStreamStartCB stream_start;
    ReorderBufferStreamStopCB stream_stop;
    ReorderBufferStreamAbortCB stream_abort;
    ReorderBufferStreamPrepareCB stream_prepare;
    ReorderBufferStreamCommitCB stream_commit;
    ReorderBufferStreamChangeCB stream_change;
    ReorderBufferStreamMessageCB stream_message;
    ReorderBufferStreamTruncateCB stream_truncate;
 
    /*
     * Callback to be called when updating progress during sending data of a
     * transaction (and its subtransactions) to the output plugin.
     */
    ReorderBufferUpdateProgressTxnCB update_progress_txn;
 
    /*
     * Pointer that will be passed untouched to the callbacks.
     */
    void       *private_data;
 
    /*
     * Saved output plugin option
     */
    bool        output_rewrites;
 
    /*
     * Private memory context.
     */
    MemoryContext context;
 
    /*
     * Memory contexts for specific types objects
     */
    MemoryContext change_context;
    MemoryContext txn_context;
    MemoryContext tup_context;
 
    XLogRecPtr  current_restart_decoding_lsn;
 
    /* buffer for disk<->memory conversions */
    char       *outbuf;
    Size        outbufsize;
 
    /* memory accounting */
    Size        size;
 
    /* Max-heap for sizes of all top-level and sub transactions */
    pairingheap *txn_heap;
 
    /*
     * Statistics about transactions spilled to disk.
     *
     * A single transaction may be spilled repeatedly, which is why we keep
     * two different counters. For spilling, the transaction counter includes
     * both toplevel transactions and subtransactions.
     */
    int64       spillTxns;      /* number of transactions spilled to disk */
    int64       spillCount;     /* spill-to-disk invocation counter */
    int64       spillBytes;     /* amount of data spilled to disk */
 
    /* Statistics about transactions streamed to the decoding output plugin */
    int64       streamTxns;     /* number of transactions streamed */
    int64       streamCount;    /* streaming invocation counter */
    int64       streamBytes;    /* amount of data decoded */
 
    /* Number of times the logical_decoding_work_mem limit has been reached */
    int64       memExceededCount;
 
    /*
     * Statistics about all the transactions sent to the decoding output
     * plugin
     */
    int64       totalTxns;      /* total number of transactions sent */
    int64       totalBytes;     /* total amount of data decoded */
};
 
 
extern ReorderBuffer *ReorderBufferAllocate(void);
extern void ReorderBufferFree(ReorderBuffer *rb);
 
extern HeapTuple ReorderBufferAllocTupleBuf(ReorderBuffer *rb, Size tuple_len);
extern void ReorderBufferFreeTupleBuf(HeapTuple tuple);
 
extern ReorderBufferChange *ReorderBufferAllocChange(ReorderBuffer *rb);
extern void ReorderBufferFreeChange(ReorderBuffer *rb,
                                    ReorderBufferChange *change, bool upd_mem);
 
extern Oid *ReorderBufferAllocRelids(ReorderBuffer *rb, int nrelids);
extern void ReorderBufferFreeRelids(ReorderBuffer *rb, Oid *relids);
 
extern void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid,
                                     XLogRecPtr lsn, ReorderBufferChange *change,
                                     bool toast_insert);
extern void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid,
                                      Snapshot snap, XLogRecPtr lsn,
                                      bool transactional, const char *prefix,
                                      Size message_size, const char *message);
extern void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
                                XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
                                TimestampTz commit_time, ReplOriginId origin_id, XLogRecPtr origin_lsn);
extern void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
                                        XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
                                        XLogRecPtr two_phase_at,
                                        TimestampTz commit_time,
                                        ReplOriginId origin_id, XLogRecPtr origin_lsn,
                                        char *gid, bool is_commit);
extern void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid,
                                     TransactionId subxid, XLogRecPtr lsn);
extern void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid,
                                     TransactionId subxid, XLogRecPtr commit_lsn,
                                     XLogRecPtr end_lsn);
extern void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
                               TimestampTz abort_time);
extern void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid);
extern void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn);
extern void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn);
 
extern void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid,
                                         XLogRecPtr lsn, Snapshot snap);
extern void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid,
                                     XLogRecPtr lsn, Snapshot snap);
extern void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid,
                                         XLogRecPtr lsn, CommandId cid);
extern void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid,
                                         XLogRecPtr lsn, RelFileLocator locator,
                                         ItemPointerData tid,
                                         CommandId cmin, CommandId cmax, CommandId combocid);
extern void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
                                          Size nmsgs, SharedInvalidationMessage *msgs);
extern void ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid,
                                                     XLogRecPtr lsn, Size nmsgs,
                                                     SharedInvalidationMessage *msgs);
extern void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations,
                                               SharedInvalidationMessage *invalidations);
extern void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn);
 
extern void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn);
extern bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid);
extern bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid);
 
extern bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid,
                                             XLogRecPtr prepare_lsn, XLogRecPtr end_lsn,
                                             TimestampTz prepare_time,
                                             ReplOriginId origin_id, XLogRecPtr origin_lsn);
extern void ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid);
extern void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, char *gid);
extern ReorderBufferTXN *ReorderBufferGetOldestTXN(ReorderBuffer *rb);
extern TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb);
extern TransactionId *ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb);
 
extern void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr);
 
extern uint32 ReorderBufferGetInvalidations(ReorderBuffer *rb,
                                            TransactionId xid,
                                            SharedInvalidationMessage **msgs);
 
extern void StartupReorderBuffer(void);
 
#endif

◆ rbtxn_get_toptxn

#define rbtxn_get_toptxn ( txn )

Value:

( \
	rbtxn_is_subtxn(txn) ? (txn)->toptxn : (txn) \
)

Definition at line 288 of file reorderbuffer.h.

290 : (txn) \

291)

◆ RBTXN_HAS_CATALOG_CHANGES

#define RBTXN_HAS_CATALOG_CHANGES 0x0001

Definition at line 167 of file reorderbuffer.h.

◆ rbtxn_has_catalog_changes

#define rbtxn_has_catalog_changes ( txn )

Value:

( \
     ((txn)->txn_flags & RBTXN_HAS_CATALOG_CHANGES) != 0 \
)

Definition at line 184 of file reorderbuffer.h.

◆ RBTXN_HAS_PARTIAL_CHANGE

#define RBTXN_HAS_PARTIAL_CHANGE 0x0020

Definition at line 172 of file reorderbuffer.h.

◆ rbtxn_has_partial_change

#define rbtxn_has_partial_change ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_HAS_PARTIAL_CHANGE) != 0 \
)

Definition at line 208 of file reorderbuffer.h.

◆ RBTXN_HAS_STREAMABLE_CHANGE

#define RBTXN_HAS_STREAMABLE_CHANGE 0x0100

Definition at line 175 of file reorderbuffer.h.

◆ rbtxn_has_streamable_change

#define rbtxn_has_streamable_change ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_HAS_STREAMABLE_CHANGE) != 0 \
)

Definition at line 214 of file reorderbuffer.h.

◆ RBTXN_IS_ABORTED

#define RBTXN_IS_ABORTED 0x0800

Definition at line 178 of file reorderbuffer.h.

◆ rbtxn_is_aborted

#define rbtxn_is_aborted ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_IS_ABORTED) != 0 \
)

Definition at line 258 of file reorderbuffer.h.

◆ RBTXN_IS_COMMITTED

#define RBTXN_IS_COMMITTED 0x0400

Definition at line 177 of file reorderbuffer.h.

◆ rbtxn_is_committed

#define rbtxn_is_committed ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_IS_COMMITTED) != 0 \
)

Definition at line 252 of file reorderbuffer.h.

◆ rbtxn_is_known_subxact

#define rbtxn_is_known_subxact ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_IS_SUBXACT) != 0 \
)

Definition at line 190 of file reorderbuffer.h.

◆ RBTXN_IS_PREPARED

#define RBTXN_IS_PREPARED 0x0040

Definition at line 173 of file reorderbuffer.h.

◆ rbtxn_is_prepared

#define rbtxn_is_prepared ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_IS_PREPARED) != 0 \
)

Definition at line 240 of file reorderbuffer.h.

◆ RBTXN_IS_SERIALIZED

#define RBTXN_IS_SERIALIZED 0x0004

Definition at line 169 of file reorderbuffer.h.

◆ rbtxn_is_serialized

#define rbtxn_is_serialized ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_IS_SERIALIZED) != 0 \
)

Definition at line 196 of file reorderbuffer.h.

◆ RBTXN_IS_SERIALIZED_CLEAR

#define RBTXN_IS_SERIALIZED_CLEAR 0x0008

Definition at line 170 of file reorderbuffer.h.

◆ rbtxn_is_serialized_clear

#define rbtxn_is_serialized_clear ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_IS_SERIALIZED_CLEAR) != 0 \
)

Definition at line 202 of file reorderbuffer.h.

◆ RBTXN_IS_STREAMED

#define RBTXN_IS_STREAMED 0x0010

Definition at line 171 of file reorderbuffer.h.

◆ rbtxn_is_streamed

#define rbtxn_is_streamed ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_IS_STREAMED) != 0 \
)

Definition at line 228 of file reorderbuffer.h.

◆ rbtxn_is_subtxn

#define rbtxn_is_subtxn ( txn )

Value:

( \
    (txn)->toptxn != NULL \
)

Definition at line 282 of file reorderbuffer.h.

◆ RBTXN_IS_SUBXACT

#define RBTXN_IS_SUBXACT 0x0002

Definition at line 168 of file reorderbuffer.h.

◆ rbtxn_is_toptxn

#define rbtxn_is_toptxn ( txn )

Value:

( \
    (txn)->toptxn == NULL \
)

Definition at line 276 of file reorderbuffer.h.

◆ RBTXN_PREPARE_STATUS_MASK

#define RBTXN_PREPARE_STATUS_MASK (RBTXN_IS_PREPARED | RBTXN_SKIPPED_PREPARE | RBTXN_SENT_PREPARE)

Definition at line 181 of file reorderbuffer.h.

◆ RBTXN_SENT_PREPARE

#define RBTXN_SENT_PREPARE 0x0200

Definition at line 176 of file reorderbuffer.h.

◆ rbtxn_sent_prepare

#define rbtxn_sent_prepare ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_SENT_PREPARE) != 0 \
)

Definition at line 246 of file reorderbuffer.h.

◆ rbtxn_skip_prepared

#define rbtxn_skip_prepared ( txn )

Value:

( \
    ((txn)->txn_flags & RBTXN_SKIPPED_PREPARE) != 0 \
)

Definition at line 264 of file reorderbuffer.h.

◆ RBTXN_SKIPPED_PREPARE

#define RBTXN_SKIPPED_PREPARE 0x0080

Definition at line 174 of file reorderbuffer.h.

Typedef Documentation

◆ ReorderBuffer

typedef struct ReorderBuffer ReorderBuffer

Definition at line 471 of file reorderbuffer.h.

◆ ReorderBufferApplyChangeCB

typedef void(* ReorderBufferApplyChangeCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)

Definition at line 474 of file reorderbuffer.h.

◆ ReorderBufferApplyTruncateCB

typedef void(* ReorderBufferApplyTruncateCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)

Definition at line 480 of file reorderbuffer.h.

◆ ReorderBufferBeginCB

typedef void(* ReorderBufferBeginCB) (ReorderBuffer *rb, ReorderBufferTXN *txn)

Definition at line 487 of file reorderbuffer.h.

◆ ReorderBufferBeginPrepareCB

typedef void(* ReorderBufferBeginPrepareCB) (ReorderBuffer *rb, ReorderBufferTXN *txn)

Definition at line 504 of file reorderbuffer.h.

◆ ReorderBufferChange

typedef struct ReorderBufferChange ReorderBufferChange

◆ ReorderBufferChangeType

typedef enum ReorderBufferChangeType ReorderBufferChangeType

◆ ReorderBufferCommitCB

typedef void(* ReorderBufferCommitCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)

Definition at line 491 of file reorderbuffer.h.

◆ ReorderBufferCommitPreparedCB

typedef void(* ReorderBufferCommitPreparedCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)

Definition at line 513 of file reorderbuffer.h.

◆ ReorderBufferMessageCB

typedef void(* ReorderBufferMessageCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message)

Definition at line 496 of file reorderbuffer.h.

◆ ReorderBufferPrepareCB

typedef void(* ReorderBufferPrepareCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)

Definition at line 508 of file reorderbuffer.h.

◆ ReorderBufferRollbackPreparedCB

typedef void(* ReorderBufferRollbackPreparedCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)

Definition at line 518 of file reorderbuffer.h.

◆ ReorderBufferStreamAbortCB

typedef void(* ReorderBufferStreamAbortCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr abort_lsn)

Definition at line 534 of file reorderbuffer.h.

◆ ReorderBufferStreamChangeCB

typedef void(* ReorderBufferStreamChangeCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)

Definition at line 549 of file reorderbuffer.h.

◆ ReorderBufferStreamCommitCB

typedef void(* ReorderBufferStreamCommitCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)

Definition at line 544 of file reorderbuffer.h.

◆ ReorderBufferStreamMessageCB

typedef void(* ReorderBufferStreamMessageCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message)

Definition at line 555 of file reorderbuffer.h.

◆ ReorderBufferStreamPrepareCB

typedef void(* ReorderBufferStreamPrepareCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)

Definition at line 539 of file reorderbuffer.h.

◆ ReorderBufferStreamStartCB

typedef void(* ReorderBufferStreamStartCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr first_lsn)

Definition at line 524 of file reorderbuffer.h.

◆ ReorderBufferStreamStopCB

typedef void(* ReorderBufferStreamStopCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr last_lsn)

Definition at line 529 of file reorderbuffer.h.

◆ ReorderBufferStreamTruncateCB

typedef void(* ReorderBufferStreamTruncateCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)

Definition at line 563 of file reorderbuffer.h.

◆ ReorderBufferTXN

typedef struct ReorderBufferTXN ReorderBufferTXN

◆ ReorderBufferUpdateProgressTxnCB

typedef void(* ReorderBufferUpdateProgressTxnCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr lsn)

Definition at line 570 of file reorderbuffer.h.

Enumeration Type Documentation

◆ DebugLogicalRepStreamingMode

enum DebugLogicalRepStreamingMode

Enumerator
DEBUG_LOGICAL_REP_STREAMING_BUFFERED
DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE

Definition at line 31 of file reorderbuffer.h.

{
    DEBUG_LOGICAL_REP_STREAMING_BUFFERED,
    DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE,
}           DebugLogicalRepStreamingMode;

◆ ReorderBufferChangeType

enum ReorderBufferChangeType

Enumerator
REORDER_BUFFER_CHANGE_INSERT
REORDER_BUFFER_CHANGE_UPDATE
REORDER_BUFFER_CHANGE_DELETE
REORDER_BUFFER_CHANGE_MESSAGE
REORDER_BUFFER_CHANGE_INVALIDATION
REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT
REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID
REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID
REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT
REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM
REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT
REORDER_BUFFER_CHANGE_TRUNCATE

Definition at line 50 of file reorderbuffer.h.

{
    REORDER_BUFFER_CHANGE_INSERT,
    REORDER_BUFFER_CHANGE_UPDATE,
    REORDER_BUFFER_CHANGE_DELETE,
    REORDER_BUFFER_CHANGE_MESSAGE,
    REORDER_BUFFER_CHANGE_INVALIDATION,
    REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT,
    REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID,
    REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID,
    REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT,
    REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM,
    REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT,
    REORDER_BUFFER_CHANGE_TRUNCATE,
} ReorderBufferChangeType;

Function Documentation

◆ ReorderBufferAbort()

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

extern

Definition at line 3085 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
 
    /* unknown, nothing to remove */
    if (txn == NULL)
        return;
 
    txn->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 only the inval
         * messages in this transaction.
         */
        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, fb(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBufferImmediateInvalidation(), and ReorderBufferTXNByXid().

Referenced by DecodeAbort().

◆ ReorderBufferAbortOld()

void ReorderBufferAbortOld	(	ReorderBuffer *	rb,
		TransactionId	oldestRunningXid
	)

extern

Definition at line 3131 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 DEBUG2, dlist_container, dlist_foreach_modify, elog, fb(), InvalidXLogRecPtr, rbtxn_is_streamed, ReorderBufferCleanupTXN(), TransactionIdPrecedes(), and ReorderBufferTXN::xid.

Referenced by standby_decode().

◆ ReorderBufferAddDistributedInvalidations()

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

extern

Definition at line 3581 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
    MemoryContext oldcontext;
 
    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
     * ReorderBufferAddInvalidations.
     */
    txn = rbtxn_get_toptxn(txn);
 
    Assert(nmsgs > 0);
 
    if (!rbtxn_distr_inval_overflowed(txn))
    {
        /*
         * Check the transaction has enough space for storing distributed
         * invalidation messages.
         */
        if (txn->ninvalidations_distributed + nmsgs >= MAX_DISTR_INVAL_MSG_PER_TXN)
        {
            /*
             * Mark the invalidation message as overflowed and free up the
             * messages accumulated so far.
             */
            txn->txn_flags |= RBTXN_DISTR_INVAL_OVERFLOWED;
 
            if (txn->invalidations_distributed)
            {
                pfree(txn->invalidations_distributed);
                txn->invalidations_distributed = NULL;
                txn->ninvalidations_distributed = 0;
            }
        }
        else
            ReorderBufferAccumulateInvalidations(&txn->invalidations_distributed,
                                                 &txn->ninvalidations_distributed,
                                                 msgs, nmsgs);
    }
 
    /* Queue the invalidation messages into the transaction */
    ReorderBufferQueueInvalidations(rb, xid, lsn, nmsgs, msgs);
 
    MemoryContextSwitchTo(oldcontext);
}

References Assert, fb(), ReorderBufferTXN::invalidations_distributed, MAX_DISTR_INVAL_MSG_PER_TXN, MemoryContextSwitchTo(), ReorderBufferTXN::ninvalidations_distributed, pfree(), RBTXN_DISTR_INVAL_OVERFLOWED, rbtxn_distr_inval_overflowed, rbtxn_get_toptxn, ReorderBufferAccumulateInvalidations(), ReorderBufferQueueInvalidations(), ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.

Referenced by SnapBuildDistributeSnapshotAndInval().

◆ ReorderBufferAddInvalidations()

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

extern

Definition at line 3540 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
    MemoryContext oldcontext;
 
    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);
 
    ReorderBufferAccumulateInvalidations(&txn->invalidations,
                                         &txn->ninvalidations,
                                         msgs, nmsgs);
 
    ReorderBufferQueueInvalidations(rb, xid, lsn, nmsgs, msgs);
 
    MemoryContextSwitchTo(oldcontext);
}

References Assert, fb(), ReorderBufferTXN::invalidations, MemoryContextSwitchTo(), ReorderBufferTXN::ninvalidations, rbtxn_get_toptxn, ReorderBufferAccumulateInvalidations(), ReorderBufferQueueInvalidations(), and ReorderBufferTXNByXid().

Referenced by xact_decode().

◆ ReorderBufferAddNewCommandId()

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

extern

Definition at line 3355 of file reorderbuffer.c.

{
    ReorderBufferChange *change = ReorderBufferAllocChange(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, fb(), REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, ReorderBufferAllocChange(), and ReorderBufferQueueChange().

Referenced by SnapBuildProcessNewCid().

◆ ReorderBufferAddNewTupleCids()

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

extern

Definition at line 3454 of file reorderbuffer.c.

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

References ReorderBufferChange::action, ReorderBufferChange::cmax, ReorderBufferChange::cmin, ReorderBufferChange::combocid, ReorderBufferChange::data, dlist_push_tail(), fb(), ReorderBufferChange::locator, ReorderBufferChange::lsn, ReorderBufferChange::node, ReorderBufferTXN::ntuplecids, REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID, ReorderBufferAllocChange(), ReorderBufferTXNByXid(), ReorderBufferChange::tid, ReorderBufferChange::tuplecid, ReorderBufferTXN::tuplecids, and ReorderBufferChange::txn.

Referenced by SnapBuildProcessNewCid().

◆ ReorderBufferAddSnapshot()

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

extern

Definition at line 3306 of file reorderbuffer.c.

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

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

Referenced by SnapBuildDistributeSnapshotAndInval().

◆ ReorderBufferAllocate()

ReorderBuffer * ReorderBufferAllocate ( void )

extern

Definition at line 325 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->memExceededCount = 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(), fb(), GenerationContextCreate(), HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, InvalidTransactionId, InvalidXLogRecPtr, ReorderBuffer::memExceededCount, 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().

◆ ReorderBufferAllocChange()

ReorderBufferChange * ReorderBufferAllocChange ( ReorderBuffer * rb )

extern

Definition at line 508 of file reorderbuffer.c.

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

References fb(), and MemoryContextAlloc().

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

◆ ReorderBufferAllocRelids()

Oid * ReorderBufferAllocRelids	(	ReorderBuffer *	rb,
		int	nrelids
	)

extern

Definition at line 626 of file reorderbuffer.c.

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

References fb(), and MemoryContextAlloc().

Referenced by DecodeTruncate(), and ReorderBufferRestoreChange().

◆ ReorderBufferAllocTupleBuf()

HeapTuple ReorderBufferAllocTupleBuf	(	ReorderBuffer *	rb,
		Size	tuple_len
	)

extern

Definition at line 593 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 fb(), HEAPTUPLESIZE, MemoryContextAlloc(), SizeofHeapTupleHeader, and HeapTupleData::t_data.

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

◆ ReorderBufferAssignChild()

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

extern

Definition at line 1100 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(), fb(), ReorderBufferTXN::nsubtxns, rbtxn_is_known_subxact, RBTXN_IS_SUBXACT, ReorderBufferTransferSnapToParent(), ReorderBufferTXNByXid(), and ReorderBufferTXN::subtxns.

Referenced by LogicalDecodingProcessRecord(), and ReorderBufferCommitChild().

◆ ReorderBufferCommit()

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

extern

Definition at line 2882 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 fb(), InvalidXLogRecPtr, ReorderBufferReplay(), and ReorderBufferTXNByXid().

Referenced by DecodeCommit().

◆ ReorderBufferCommitChild()

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

extern

Definition at line 1220 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 fb(), InvalidXLogRecPtr, ReorderBufferAssignChild(), and ReorderBufferTXNByXid().

Referenced by DecodeCommit(), and DecodePrepare().

◆ ReorderBufferFinishPrepared()

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

extern

Definition at line 2999 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->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 must have been marked as a prepared transaction and skipped but
         * not sent a prepare. Also, the prepare info must have been updated
         * in txn even if we skip prepare.
         */
        Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) ==
               (RBTXN_IS_PREPARED | RBTXN_SKIPPED_PREPARE));
        Assert(XLogRecPtrIsValid(txn->final_lsn));
 
        /*
         * 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->prepare_time, txn->origin_id, txn->origin_lsn);
    }
 
    txn->final_lsn = commit_lsn;
    txn->end_lsn = end_lsn;
    txn->commit_time = commit_time;
    txn->origin_id = origin_id;
    txn->origin_lsn = origin_lsn;
 
    if (is_commit)
        rb->commit_prepared(rb, txn, commit_lsn);
    else
        rb->rollback_prepared(rb, txn, prepare_end_lsn, prepare_time);
 
    /* cleanup: make sure there's no cache pollution */
    ReorderBufferExecuteInvalidations(txn->ninvalidations,
                                      txn->invalidations);
    ReorderBufferCleanupTXN(rb, txn);
}

Referenced by DecodeAbort(), and DecodeCommit().

◆ ReorderBufferForget()

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

extern

Definition at line 3178 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 only cache invalidation messages in this transaction 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, fb(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBufferImmediateInvalidation(), and ReorderBufferTXNByXid().

Referenced by DecodeCommit().

◆ ReorderBufferFree()

void ReorderBufferFree ( ReorderBuffer * rb )

extern

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

Referenced by FreeDecodingContext().

◆ ReorderBufferFreeChange()

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

extern

Definition at line 523 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)
            {
                ReorderBufferFreeTupleBuf(change->data.tp.newtuple);
                change->data.tp.newtuple = NULL;
            }
 
            if (change->data.tp.oldtuple)
            {
                ReorderBufferFreeTupleBuf(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)
            {
                ReorderBufferFreeRelids(rb, change->data.truncate.relids);
                change->data.truncate.relids = NULL;
            }
            break;
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
        case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
        case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
        case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
            break;
    }
 
    pfree(change);
}

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

◆ ReorderBufferFreeRelids()

void ReorderBufferFreeRelids	(	ReorderBuffer *	rb,
		Oid *	relids
	)

extern

Definition at line 642 of file reorderbuffer.c.

{
    pfree(relids);
}

References pfree().

Referenced by ReorderBufferFreeChange().

◆ ReorderBufferFreeTupleBuf()

void ReorderBufferFreeTupleBuf ( HeapTuple tuple )

extern

Definition at line 611 of file reorderbuffer.c.

{
    pfree(tuple);
}

References pfree().

Referenced by ReorderBufferFreeChange().

◆ ReorderBufferGetCatalogChangesXacts()

TransactionId * ReorderBufferGetCatalogChangesXacts ( ReorderBuffer * rb )

extern

Definition at line 3689 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 = palloc_array(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, dlist_iter::cur, dclist_container, dclist_count(), dclist_foreach, fb(), palloc_array, qsort, rbtxn_has_catalog_changes, ReorderBufferTXN::xid, and xidComparator().

Referenced by SnapBuildSerialize().

◆ ReorderBufferGetInvalidations()

uint32 ReorderBufferGetInvalidations	(	ReorderBuffer *	rb,
		TransactionId	xid,
		SharedInvalidationMessage **	msgs
	)

extern

Definition at line 5629 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
 
    if (txn == NULL)
        return 0;
 
    *msgs = txn->invalidations;
 
    return txn->ninvalidations;
}

References fb(), ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, and ReorderBufferTXNByXid().

Referenced by SnapBuildDistributeSnapshotAndInval().

◆ ReorderBufferGetOldestTXN()

ReorderBufferTXN * ReorderBufferGetOldestTXN ( ReorderBuffer * rb )

extern

Definition at line 1045 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(XLogRecPtrIsValid(txn->first_lsn));
    return txn;
}

References Assert, AssertTXNLsnOrder(), dlist_head_element, dlist_is_empty(), fb(), ReorderBufferTXN::first_lsn, rbtxn_is_known_subxact, and XLogRecPtrIsValid.

Referenced by SnapBuildProcessRunningXacts().

◆ ReorderBufferGetOldestXmin()

TransactionId ReorderBufferGetOldestXmin ( ReorderBuffer * rb )

extern

Definition at line 1073 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(), fb(), InvalidTransactionId, and SnapshotData::xmin.

Referenced by SnapBuildProcessRunningXacts().

◆ ReorderBufferImmediateInvalidation()

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

extern

Definition at line 3251 of file reorderbuffer.c.

{
    bool        use_subtxn = IsTransactionOrTransactionBlock();
    MemoryContext ccxt = CurrentMemoryContext;
    ResourceOwner cowner = CurrentResourceOwner;
    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();
        MemoryContextSwitchTo(ccxt);
        CurrentResourceOwner = cowner;
    }
}

References AbortCurrentTransaction(), BeginInternalSubTransaction(), CurrentMemoryContext, CurrentResourceOwner, fb(), i, IsTransactionOrTransactionBlock(), LocalExecuteInvalidationMessage(), MemoryContextSwitchTo(), and RollbackAndReleaseCurrentSubTransaction().

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

◆ ReorderBufferInvalidate()

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

extern

Definition at line 3220 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, fb(), ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, ReorderBufferImmediateInvalidation(), and ReorderBufferTXNByXid().

Referenced by DecodePrepare().

◆ ReorderBufferPrepare()

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

extern

Definition at line 2958 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
                                false);
 
    /* unknown transaction, nothing to replay */
    if (txn == NULL)
        return;
 
    /*
     * txn must have been marked as a prepared transaction and must have
     * neither been skipped nor sent a prepare. Also, the prepare info must
     * have been updated in it by now.
     */
    Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) == RBTXN_IS_PREPARED);
    Assert(XLogRecPtrIsValid(txn->final_lsn));
 
    txn->gid = pstrdup(gid);
 
    ReorderBufferReplay(txn, rb, xid, txn->final_lsn, txn->end_lsn,
                        txn->prepare_time, txn->origin_id, txn->origin_lsn);
 
    /*
     * Send a prepare if not already done so. This might occur if we have
     * detected a concurrent abort while replaying the non-streaming
     * transaction.
     */
    if (!rbtxn_sent_prepare(txn))
    {
        rb->prepare(rb, txn, txn->final_lsn);
        txn->txn_flags |= RBTXN_SENT_PREPARE;
    }
}

References Assert, ReorderBufferTXN::end_lsn, fb(), ReorderBufferTXN::final_lsn, ReorderBufferTXN::gid, InvalidXLogRecPtr, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBufferTXN::prepare_time, pstrdup(), RBTXN_IS_PREPARED, RBTXN_PREPARE_STATUS_MASK, RBTXN_SENT_PREPARE, rbtxn_sent_prepare, ReorderBufferReplay(), ReorderBufferTXNByXid(), ReorderBufferTXN::txn_flags, and XLogRecPtrIsValid.

Referenced by DecodePrepare().

◆ ReorderBufferProcessXid()

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

extern

Definition at line 3293 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 fb(), 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
	)

extern

Definition at line 811 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
    /*
     * If we have detected that the transaction is aborted while streaming the
     * previous changes or by checking its CLOG, there is no point in
     * collecting further changes for it.
     */
    if (rbtxn_is_aborted(txn))
    {
        /*
         * We don't need to update memory accounting for this change as we
         * have not added it to the queue yet.
         */
        ReorderBufferFreeChange(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(XLogRecPtrIsValid(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, dlist_push_tail(), fb(), ReorderBufferChange::lsn, ReorderBufferTXN::nentries, ReorderBufferTXN::nentries_mem, ReorderBufferChange::node, rbtxn_get_toptxn, RBTXN_HAS_STREAMABLE_CHANGE, rbtxn_is_aborted, 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(), ReorderBufferFreeChange(), ReorderBufferProcessPartialChange(), ReorderBufferTXNByXid(), ReorderBufferChange::txn, ReorderBufferTXN::txn_flags, and XLogRecPtrIsValid.

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

◆ ReorderBufferQueueMessage()

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

extern

Definition at line 874 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 = ReorderBufferAllocChange(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, ReorderBufferChange::data, fb(), InvalidTransactionId, MemoryContextSwitchTo(), ReorderBufferChange::message, ReorderBufferChange::message_size, ReorderBufferChange::msg, palloc(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, ReorderBufferChange::prefix, pstrdup(), REORDER_BUFFER_CHANGE_MESSAGE, ReorderBufferAllocChange(), 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,
		ReplOriginId	origin_id,
		XLogRecPtr	origin_lsn
	)

extern

Definition at line 2905 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->prepare_time = prepare_time;
    txn->origin_id = origin_id;
    txn->origin_lsn = origin_lsn;
 
    /* Mark this transaction as a prepared transaction */
    Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) == 0);
    txn->txn_flags |= RBTXN_IS_PREPARED;
 
    return true;
}

References Assert, ReorderBufferTXN::end_lsn, fb(), ReorderBufferTXN::final_lsn, InvalidXLogRecPtr, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBufferTXN::prepare_time, RBTXN_IS_PREPARED, RBTXN_PREPARE_STATUS_MASK, ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.

Referenced by DecodePrepare().

◆ ReorderBufferSetBaseSnapshot()

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

extern

Definition at line 3324 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(), fb(), InvalidXLogRecPtr, rbtxn_is_known_subxact, ReorderBufferTXNByXid(), and ReorderBufferTXN::toplevel_xid.

Referenced by SnapBuildCommitTxn(), and SnapBuildProcessChange().

◆ ReorderBufferSetRestartPoint()

void ReorderBufferSetRestartPoint	(	ReorderBuffer *	rb,
		XLogRecPtr	ptr
	)

extern

Definition at line 1088 of file reorderbuffer.c.

{
    rb->current_restart_decoding_lsn = ptr;
}

References fb().

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

◆ ReorderBufferSkipPrepare()

void ReorderBufferSkipPrepare	(	ReorderBuffer *	rb,
		TransactionId	xid
	)

extern

Definition at line 2937 of file reorderbuffer.c.

{
    ReorderBufferTXN *txn;
 
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false);
 
    /* unknown transaction, nothing to do */
    if (txn == NULL)
        return;
 
    /* txn must have been marked as a prepared transaction */
    Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) == RBTXN_IS_PREPARED);
    txn->txn_flags |= RBTXN_SKIPPED_PREPARE;
}

References Assert, fb(), InvalidXLogRecPtr, RBTXN_IS_PREPARED, RBTXN_PREPARE_STATUS_MASK, RBTXN_SKIPPED_PREPARE, ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.

Referenced by DecodePrepare().

◆ ReorderBufferXidHasBaseSnapshot()

bool ReorderBufferXidHasBaseSnapshot	(	ReorderBuffer *	rb,
		TransactionId	xid
	)

extern

Definition at line 3740 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, fb(), InvalidXLogRecPtr, rbtxn_is_known_subxact, ReorderBufferTXNByXid(), and ReorderBufferTXN::toplevel_xid.

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

◆ ReorderBufferXidHasCatalogChanges()

bool ReorderBufferXidHasCatalogChanges	(	ReorderBuffer *	rb,
		TransactionId	xid
	)

extern

Definition at line 3723 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 fb(), InvalidXLogRecPtr, rbtxn_has_catalog_changes, and ReorderBufferTXNByXid().

Referenced by SnapBuildXidHasCatalogChanges().

◆ ReorderBufferXidSetCatalogChanges()

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

extern

Definition at line 3651 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, dclist_push_tail(), fb(), rbtxn_get_toptxn, RBTXN_HAS_CATALOG_CHANGES, rbtxn_has_catalog_changes, rbtxn_is_subtxn, ReorderBufferTXNByXid(), and ReorderBufferTXN::txn_flags.

Referenced by SnapBuildProcessNewCid(), and xact_decode().

◆ StartupReorderBuffer()

void StartupReorderBuffer ( void )

extern

Definition at line 4939 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, true, 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(), DEBUG2, fb(), FreeDir(), PG_REPLSLOT_DIR, ReadDir(), ReorderBufferCleanupSerializedTXNs(), and ReplicationSlotValidateName().

Referenced by StartupXLOG().

Variable Documentation

◆ debug_logical_replication_streaming

PGDLLIMPORT int debug_logical_replication_streaming

extern

Definition at line 230 of file reorderbuffer.c.

Referenced by pa_send_data(), ReorderBufferCheckAndTruncateAbortedTXN(), and ReorderBufferCheckMemoryLimit().

◆ logical_decoding_work_mem

PGDLLIMPORT int logical_decoding_work_mem

extern

Definition at line 226 of file reorderbuffer.c.

Referenced by ReorderBufferCheckMemoryLimit().

Data Structures

Macros

Typedefs

Enumerations

Functions

Variables

Macro Definition Documentation

◆ PG_LOGICAL_DIR

◆ PG_LOGICAL_MAPPINGS_DIR

◆ PG_LOGICAL_SNAPSHOTS_DIR

◆ RBTXN_DISTR_INVAL_OVERFLOWED

◆ rbtxn_distr_inval_overflowed

◆ rbtxn_get_toptxn

◆ RBTXN_HAS_CATALOG_CHANGES

◆ rbtxn_has_catalog_changes

◆ RBTXN_HAS_PARTIAL_CHANGE

◆ rbtxn_has_partial_change

◆ RBTXN_HAS_STREAMABLE_CHANGE

◆ rbtxn_has_streamable_change

◆ RBTXN_IS_ABORTED

◆ rbtxn_is_aborted

◆ RBTXN_IS_COMMITTED

◆ rbtxn_is_committed

◆ rbtxn_is_known_subxact

◆ RBTXN_IS_PREPARED

◆ rbtxn_is_prepared

◆ RBTXN_IS_SERIALIZED

◆ rbtxn_is_serialized

◆ RBTXN_IS_SERIALIZED_CLEAR

◆ rbtxn_is_serialized_clear

◆ RBTXN_IS_STREAMED

◆ rbtxn_is_streamed

◆ rbtxn_is_subtxn

◆ RBTXN_IS_SUBXACT

◆ rbtxn_is_toptxn

◆ RBTXN_PREPARE_STATUS_MASK

◆ RBTXN_SENT_PREPARE

◆ rbtxn_sent_prepare

◆ rbtxn_skip_prepared

◆ RBTXN_SKIPPED_PREPARE

Typedef Documentation

◆ ReorderBuffer

◆ ReorderBufferApplyChangeCB

◆ ReorderBufferApplyTruncateCB

◆ ReorderBufferBeginCB

◆ ReorderBufferBeginPrepareCB

◆ ReorderBufferChange

◆ ReorderBufferChangeType

◆ ReorderBufferCommitCB

◆ ReorderBufferCommitPreparedCB

◆ ReorderBufferMessageCB

◆ ReorderBufferPrepareCB

◆ ReorderBufferRollbackPreparedCB

◆ ReorderBufferStreamAbortCB

◆ ReorderBufferStreamChangeCB

◆ ReorderBufferStreamCommitCB

◆ ReorderBufferStreamMessageCB

◆ ReorderBufferStreamPrepareCB

◆ ReorderBufferStreamStartCB

◆ ReorderBufferStreamStopCB

◆ ReorderBufferStreamTruncateCB

◆ ReorderBufferTXN

◆ ReorderBufferUpdateProgressTxnCB

Enumeration Type Documentation

◆ DebugLogicalRepStreamingMode

◆ ReorderBufferChangeType

Function Documentation

◆ ReorderBufferAbort()

◆ ReorderBufferAbortOld()

◆ ReorderBufferAddDistributedInvalidations()

◆ ReorderBufferAddInvalidations()

◆ ReorderBufferAddNewCommandId()

◆ ReorderBufferAddNewTupleCids()

◆ ReorderBufferAddSnapshot()

◆ ReorderBufferAllocate()

◆ ReorderBufferAllocChange()

◆ ReorderBufferAllocRelids()

◆ ReorderBufferAllocTupleBuf()

◆ ReorderBufferAssignChild()

◆ ReorderBufferCommit()