reorderbuffer.h File Reference

#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, RepOriginId origin_id, XLogRecPtr origin_lsn)
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	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, RepOriginId 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.

rbtxn_get_toptxn

#define rbtxn_get_toptxn

(

txn

)

Value:

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

Definition at line 288 of file reorderbuffer.h.

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
	)

Definition at line 3087 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, ReorderBufferTXN::final_lsn, ReorderBufferTXN::invalidations, InvalidXLogRecPtr, ReorderBufferTXN::ninvalidations, rbtxn_is_streamed, ReorderBufferCleanupTXN(), ReorderBufferImmediateInvalidation(), ReorderBufferTXNByXid(), and ReorderBuffer::stream_abort.

Referenced by DecodeAbort().

ReorderBufferAbortOld()

void ReorderBufferAbortOld	(	ReorderBuffer *	rb,
		TransactionId	oldestRunningXid
	)

Definition at line 3133 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().

ReorderBufferAddDistributedInvalidations()

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

Definition at line 3585 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(), ReorderBuffer::context, 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
	)

Definition at line 3544 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(), ReorderBuffer::context, 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
	)

Definition at line 3357 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, 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
	)

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

Definition at line 3308 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, REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT, ReorderBufferAllocChange(), ReorderBufferQueueChange(), and ReorderBufferChange::snapshot.

Referenced by SnapBuildDistributeSnapshotAndInval().

ReorderBufferAllocate()

ReorderBuffer * ReorderBufferAllocate

(

void

)

Definition at line 324 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(), HASHCTL::entrysize, GenerationContextCreate(), HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASHCTL::hcxt, InvalidTransactionId, InvalidXLogRecPtr, HASHCTL::keysize, 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

)

Definition at line 507 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(), ReorderBufferAddNewCommandId(), ReorderBufferAddNewTupleCids(), ReorderBufferAddSnapshot(), ReorderBufferQueueInvalidations(), ReorderBufferQueueMessage(), and ReorderBufferRestoreChange().

ReorderBufferAllocRelids()

Oid * ReorderBufferAllocRelids	(	ReorderBuffer *	rb,
		int	nrelids
	)

Definition at line 625 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().

ReorderBufferAllocTupleBuf()

HeapTuple ReorderBufferAllocTupleBuf	(	ReorderBuffer *	rb,
		Size	tuple_len
	)

Definition at line 592 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().

ReorderBufferAssignChild()

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

Definition at line 1099 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().

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 2884 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 1219 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().

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 3001 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);
}

References Assert(), ReorderBuffer::commit_prepared, ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, ReorderBufferTXN::final_lsn, ReorderBufferTXN::gid, ReorderBufferTXN::invalidations, ReorderBufferTXN::ninvalidations, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBufferTXN::prepare_time, pstrdup(), RBTXN_IS_PREPARED, RBTXN_PREPARE_STATUS_MASK, RBTXN_SKIPPED_PREPARE, ReorderBufferCleanupTXN(), ReorderBufferExecuteInvalidations(), ReorderBufferReplay(), ReorderBufferTXNByXid(), ReorderBuffer::rollback_prepared, ReorderBufferTXN::txn_flags, and XLogRecPtrIsValid.

Referenced by DecodeAbort(), and DecodeCommit().

ReorderBufferForget()

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

Definition at line 3180 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, 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 417 of file reorderbuffer.c.

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

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

Referenced by FreeDecodingContext().

ReorderBufferFreeChange()

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

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

References ReorderBufferChange::action, ReorderBufferChange::data, ReorderBufferChange::inval, ReorderBufferChange::invalidations, ReorderBufferChange::message, ReorderBufferChange::msg, ReorderBufferChange::newtuple, ReorderBufferChange::oldtuple, pfree(), ReorderBufferChange::prefix, ReorderBufferChange::relids, 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(), ReorderBufferFreeRelids(), ReorderBufferFreeSnap(), ReorderBufferFreeTupleBuf(), ReorderBufferChange::snapshot, ReorderBufferChange::tp, and ReorderBufferChange::truncate.

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

ReorderBufferFreeRelids()

void ReorderBufferFreeRelids	(	ReorderBuffer *	rb,
		Oid *	relids
	)

Definition at line 641 of file reorderbuffer.c.

{
    pfree(relids);
}

References pfree().

Referenced by ReorderBufferFreeChange().

ReorderBufferFreeTupleBuf()

void ReorderBufferFreeTupleBuf

(

HeapTuple

tuple

)

Definition at line 610 of file reorderbuffer.c.

{
    pfree(tuple);
}

References pfree().

Referenced by ReorderBufferFreeChange().

ReorderBufferGetCatalogChangesXacts()

TransactionId * ReorderBufferGetCatalogChangesXacts

(

ReorderBuffer *

rb

)

Definition at line 3693 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().

ReorderBufferGetInvalidations()

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

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

Referenced by SnapBuildDistributeSnapshotAndInval().

ReorderBufferGetOldestTXN()

ReorderBufferTXN * ReorderBufferGetOldestTXN

(

ReorderBuffer *

rb

)

Definition at line 1044 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(), ReorderBufferTXN::first_lsn, rbtxn_is_known_subxact, ReorderBuffer::toplevel_by_lsn, and XLogRecPtrIsValid.

Referenced by SnapBuildProcessRunningXacts().

ReorderBufferGetOldestXmin()

TransactionId ReorderBufferGetOldestXmin

(

ReorderBuffer *

rb

)

Definition at line 1072 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().

ReorderBufferImmediateInvalidation()

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

Definition at line 3253 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, i, IsTransactionOrTransactionBlock(), LocalExecuteInvalidationMessage(), MemoryContextSwitchTo(), and RollbackAndReleaseCurrentSubTransaction().

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

ReorderBufferInvalidate()

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

Definition at line 3222 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().

ReorderBufferPrepare()

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

Definition at line 2960 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, ReorderBufferTXN::final_lsn, ReorderBufferTXN::gid, InvalidXLogRecPtr, ReorderBufferTXN::origin_id, ReorderBufferTXN::origin_lsn, ReorderBuffer::prepare, 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
	)

Definition at line 3295 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 810 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(), 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
	)

Definition at line 873 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(), ReorderBuffer::context, ReorderBufferChange::data, InvalidTransactionId, MemoryContextSwitchTo(), ReorderBufferChange::message, ReorderBuffer::message, ReorderBufferChange::message_size, ReorderBufferChange::msg, palloc(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, ReorderBufferChange::prefix, pstrdup(), REORDER_BUFFER_CHANGE_MESSAGE, 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,
		RepOriginId	origin_id,
		XLogRecPtr	origin_lsn
	)

Definition at line 2907 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, 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
	)

Definition at line 3326 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 1087 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 2939 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(), 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
	)

Definition at line 3745 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(), SnapBuildDistributeSnapshotAndInval(), and SnapBuildProcessChange().

ReorderBufferXidHasCatalogChanges()

bool ReorderBufferXidHasCatalogChanges	(	ReorderBuffer *	rb,
		TransactionId	xid
	)

Definition at line 3728 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 3655 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 SnapBuildProcessNewCid(), and xact_decode().

StartupReorderBuffer()

void StartupReorderBuffer

(

void

)

Definition at line 4944 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(), dirent::d_name, DEBUG2, 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 229 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 225 of file reorderbuffer.c.

Referenced by ReorderBufferCheckMemoryLimit().