184#define IsSpecInsert(action) \
186 ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT) \
188#define IsSpecConfirmOrAbort(action) \
190 (((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM) || \
191 ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT)) \
193#define IsInsertOrUpdate(action) \
195 (((action) == REORDER_BUFFER_CHANGE_INSERT) || \
196 ((action) == REORDER_BUFFER_CHANGE_UPDATE) || \
197 ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT)) \
302 bool addition,
Size sz);
325 memset(&hash_ctl, 0,
sizeof(hash_ctl));
454 if (txn->
gid != NULL)
608 alloc_len =
sizeof(
Oid) * nrelids;
632 bool *is_new,
XLogRecPtr lsn,
bool create_as_top)
706 Assert(!create || txn != NULL);
754 toptxn->
txn_flags &= ~RBTXN_HAS_PARTIAL_CHANGE;
765 toptxn->
txn_flags &= ~RBTXN_HAS_PARTIAL_CHANGE;
853 bool transactional,
const char *prefix,
854 Size message_size,
const char *message)
886 volatile Snapshot snapshot_now = snap;
898 rb->
message(rb, txn, lsn,
false, prefix, message_size, message);
922#ifdef USE_ASSERT_CHECKING
955 Assert(prev_first_lsn < cur_txn->first_lsn);
975 Assert(prev_base_snap_lsn < cur_txn->base_snapshot_lsn);
993#ifdef USE_ASSERT_CHECKING
1010 Assert(prev_lsn <= cur_change->lsn);
1012 prev_lsn = cur_change->
lsn;
1247 else if (pos_a == pos_b)
1302 state->nr_txns = nr_txns;
1305 for (off = 0; off <
state->nr_txns; off++)
1307 state->entries[off].file.vfd = -1;
1308 state->entries[off].segno = 0;
1317 *iter_state =
state;
1336 &
state->entries[off].segno);
1342 state->entries[off].lsn = cur_change->
lsn;
1343 state->entries[off].change = cur_change;
1344 state->entries[off].txn = txn;
1365 &
state->entries[off].file,
1366 &
state->entries[off].segno);
1371 state->entries[off].lsn = cur_change->
lsn;
1372 state->entries[off].change = cur_change;
1373 state->entries[off].txn = cur_txn;
1397 if (
state->heap->bh_size == 0)
1401 entry = &
state->entries[off];
1427 state->entries[off].lsn = next_change->
lsn;
1428 state->entries[off].change = next_change;
1451 &
state->entries[off].segno))
1458 elog(
DEBUG2,
"restored %u/%u changes from disk",
1464 state->entries[off].lsn = next_change->
lsn;
1465 state->entries[off].change = next_change;
1487 for (off = 0; off <
state->nr_txns; off++)
1489 if (
state->entries[off].file.vfd != -1)
2009 int nrelations,
Relation *relations,
2048 if (snapshot_now->
copied)
2076 if (specinsert != NULL)
2120 volatile bool stream_started =
false;
2144 int changes_count = 0;
2182 stream_started =
true;
2193 prev_lsn = change->
lsn;
2202 curtxn = change->
txn;
2215 if (specinsert == NULL)
2216 elog(
ERROR,
"invalid ordering of speculative insertion changes");
2218 change = specinsert;
2247 elog(
ERROR,
"could not map filenumber \"%s\" to relation OID",
2254 elog(
ERROR,
"could not open relation with OID %u (for filenumber \"%s\")",
2275 if (relation->
rd_rel->relkind == RELKIND_SEQUENCE)
2317 if (specinsert != NULL)
2346 if (specinsert != NULL)
2354 specinsert = change;
2367 if (specinsert != NULL)
2391 for (
i = 0;
i < nrelids;
i++)
2399 elog(
ERROR,
"could not open relation with OID %u", relid);
2404 relations[nrelations++] = rel;
2412 for (
i = 0;
i < nrelations;
i++)
2432 if (snapshot_now->
copied)
2463 if (command_id < change->
data.command_id)
2467 if (!snapshot_now->
copied)
2474 snapshot_now->
curcid = command_id;
2483 elog(
ERROR,
"tuplecid value in changequeue");
2497#define CHANGES_THRESHOLD 100
2536 stream_started =
false;
2546 rb->
prepare(rb, txn, commit_lsn);
2548 rb->
commit(rb, txn, commit_lsn);
2553 elog(
ERROR,
"output plugin used XID %u",
2562 else if (snapshot_now->
copied)
2641 if (errdata->
sqlerrcode == ERRCODE_TRANSACTION_ROLLBACK &&
2655 command_id, prev_lsn,
2755 origin_id, origin_lsn);
2850 XLogRecPtr origin_lsn,
char *gid,
bool is_commit)
2866 prepare_end_lsn = txn->
end_lsn;
2880 if ((txn->
final_lsn < two_phase_at) && is_commit)
3113 for (
i = 0;
i < ninvalidations;
i++)
3224 bool addition,
Size sz)
3392 for (
i = 0;
i < nmsgs;
i++)
3459 xids[xcnt++] = txn->
xid;
3596 Size largest_size = 0;
3611 if ((largest == NULL || txn->
total_size > largest_size) &&
3719 elog(
DEBUG2,
"spill %u changes in XID %u to disk",
3761 O_CREAT | O_WRONLY | O_APPEND |
PG_BINARY);
3766 errmsg(
"could not open file \"%s\": %m", path)));
3836 oldlen = oldtup->
t_len;
3843 newlen = newtup->
t_len;
3888 memcpy(
data, &prefix_size,
sizeof(
Size));
3892 data += prefix_size;
3991 int save_errno = errno;
3996 errno = save_errno ? save_errno : ENOSPC;
3999 errmsg(
"could not write to data file for XID %u: %m",
4060 bool txn_is_streamed;
4195 oldlen = oldtup->
t_len;
4202 newlen = newtup->
t_len;
4311 if (*
fd < 0 && errno == ENOENT)
4320 errmsg(
"could not open file \"%s\": %m",
4332 file->
curOffset, WAIT_EVENT_REORDER_BUFFER_READ);
4342 else if (readBytes < 0)
4345 errmsg(
"could not read from reorderbuffer spill file: %m")));
4349 errmsg(
"could not read from reorderbuffer spill file: read %d instead of %u bytes",
4365 WAIT_EVENT_REORDER_BUFFER_READ);
4370 errmsg(
"could not read from reorderbuffer spill file: %m")));
4374 errmsg(
"could not read from reorderbuffer spill file: read %d instead of %u bytes",
4475 memcpy(&prefix_size,
data,
sizeof(
Size));
4481 data += prefix_size;
4586 if (unlink(path) != 0 && errno != ENOENT)
4589 errmsg(
"could not remove file \"%s\": %m", path)));
4602 struct stat statbuf;
4615 if (strncmp(spill_de->
d_name,
"xid", 3) == 0)
4621 if (unlink(path) != 0)
4624 errmsg(
"could not remove file \"%s\" during removal of %s/%s/xid*: %m",
4658 struct dirent *logical_de;
4663 if (strcmp(logical_de->
d_name,
".") == 0 ||
4664 strcmp(logical_de->
d_name,
"..") == 0)
4746 elog(
ERROR,
"got sequence entry %d for toast chunk %u instead of seq 0",
4747 chunk_seq, chunk_id);
4750 elog(
ERROR,
"got sequence entry %d for toast chunk %u instead of seq %d",
4763 elog(
ERROR,
"unexpected type of toast chunk");
4765 ent->
size += chunksize;
4832 elog(
ERROR,
"could not open toast relation with OID %u (base relation \"%s\")",
4846 for (natt = 0; natt < desc->
natts; natt++)
4855 struct varlena *new_datum = NULL;
4856 struct varlena *reconstructed;
4861 if (attr->attnum < 0)
4864 if (attr->attisdropped)
4868 if (attr->attlen != -1)
4920 memcpy(
VARDATA(reconstructed) + data_done,
4933 memset(&redirect_pointer, 0,
sizeof(redirect_pointer));
4934 redirect_pointer.
pointer = reconstructed;
4938 sizeof(redirect_pointer));
4961 for (natt = 0; natt < desc->
natts; natt++)
5062 elog(
DEBUG3,
"mapping: node: %u/%u/%u tid: %u/%u cmin: %u, cmax: %u",
5094 errmsg(
"could not open file \"%s\": %m", path)));
5114 errmsg(
"could not read file \"%s\": %m",
5116 else if (readBytes == 0)
5121 errmsg(
"could not read from file \"%s\": read %d instead of %d bytes",
5166 errmsg(
"could not close file \"%s\": %m", path)));
5176 return bsearch(&xid, xip, num,
5200 struct dirent *mapping_de;
5217 if (strcmp(mapping_de->
d_name,
".") == 0 ||
5218 strcmp(mapping_de->
d_name,
"..") == 0)
5222 if (strncmp(mapping_de->
d_name,
"map-", 4) != 0)
5226 &f_dboid, &f_relid, &f_hi, &f_lo,
5227 &f_mapped_xid, &f_create_xid) != 6)
5230 f_lsn = ((
uint64) f_hi) << 32 | f_lo;
5233 if (f_dboid != dboid)
5237 if (f_relid != relid)
5259 foreach(file, files)
5284 bool updated_mapping =
false;
5298 memset(&
key, 0,
sizeof(
key));
5325 if (ent == NULL && !updated_mapping)
5329 updated_mapping =
true;
5332 else if (ent == NULL)
void binaryheap_build(binaryheap *heap)
void binaryheap_replace_first(binaryheap *heap, bh_node_type d)
bh_node_type binaryheap_first(binaryheap *heap)
bh_node_type binaryheap_remove_first(binaryheap *heap)
void binaryheap_free(binaryheap *heap)
void binaryheap_add_unordered(binaryheap *heap, bh_node_type d)
binaryheap * binaryheap_allocate(int capacity, binaryheap_comparator compare, void *arg)
static void cleanup(void)
#define BufferIsLocal(buffer)
void BufferGetTag(Buffer buffer, RelFileLocator *rlocator, ForkNumber *forknum, BlockNumber *blknum)
#define Assert(condition)
#define FLEXIBLE_ARRAY_MEMBER
bool IsToastRelation(Relation relation)
bool IsSharedRelation(Oid relationId)
#define INDIRECT_POINTER_SIZE
#define VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr)
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
void hash_destroy(HTAB *hashp)
void * hash_seq_search(HASH_SEQ_STATUS *status)
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
void FreeErrorData(ErrorData *edata)
int errcode_for_file_access(void)
ErrorData * CopyErrorData(void)
void FlushErrorState(void)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
int CloseTransientFile(int fd)
struct dirent * ReadDirExtended(DIR *dir, const char *dirname, int elevel)
void FileClose(File file)
File PathNameOpenFile(const char *fileName, int fileFlags)
DIR * AllocateDir(const char *dirname)
struct dirent * ReadDir(DIR *dir, const char *dirname)
int OpenTransientFile(const char *fileName, int fileFlags)
static ssize_t FileRead(File file, void *buffer, size_t amount, off_t offset, uint32 wait_event_info)
MemoryContext GenerationContextCreate(MemoryContext parent, const char *name, Size minContextSize, Size initBlockSize, Size maxBlockSize)
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, const Datum *values, const bool *isnull)
void heap_deform_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *values, bool *isnull)
HeapTupleData * HeapTuple
struct HeapTupleData HeapTupleData
HeapTupleHeaderData * HeapTupleHeader
#define SizeofHeapTupleHeader
static Datum fastgetattr(HeapTuple tup, int attnum, TupleDesc tupleDesc, bool *isnull)
static dlist_node * dlist_pop_head_node(dlist_head *head)
#define dlist_foreach(iter, lhead)
static void dlist_init(dlist_head *head)
#define dclist_container(type, membername, ptr)
static bool dlist_has_next(const dlist_head *head, const dlist_node *node)
static void dclist_push_tail(dclist_head *head, dlist_node *node)
static void dlist_insert_before(dlist_node *before, dlist_node *node)
#define dlist_head_element(type, membername, lhead)
static dlist_node * dlist_next_node(dlist_head *head, dlist_node *node)
static void dlist_delete(dlist_node *node)
static uint32 dclist_count(const dclist_head *head)
#define dlist_foreach_modify(iter, lhead)
static bool dlist_is_empty(const dlist_head *head)
static void dlist_push_tail(dlist_head *head, dlist_node *node)
static void dclist_delete_from(dclist_head *head, dlist_node *node)
static void dclist_init(dclist_head *head)
#define dlist_container(type, membername, ptr)
#define dclist_foreach(iter, lhead)
static int pg_cmp_u64(uint64 a, uint64 b)
void LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
if(TABLE==NULL||TABLE_index==NULL)
static OffsetNumber ItemPointerGetOffsetNumber(const ItemPointerData *pointer)
static BlockNumber ItemPointerGetBlockNumber(const ItemPointerData *pointer)
static void ItemPointerCopy(const ItemPointerData *fromPointer, ItemPointerData *toPointer)
List * lappend(List *list, void *datum)
void list_sort(List *list, list_sort_comparator cmp)
void UpdateDecodingStats(LogicalDecodingContext *ctx)
void * MemoryContextAlloc(MemoryContext context, Size size)
void * MemoryContextAllocZero(MemoryContext context, Size size)
char * pstrdup(const char *in)
void * repalloc(void *pointer, Size size)
void pfree(void *pointer)
void * palloc0(Size size)
MemoryContext CurrentMemoryContext
void MemoryContextDelete(MemoryContext context)
#define AllocSetContextCreate
#define ALLOCSET_DEFAULT_SIZES
#define SLAB_DEFAULT_BLOCK_SIZE
#define CHECK_FOR_INTERRUPTS()
void pairingheap_remove(pairingheap *heap, pairingheap_node *node)
void pairingheap_add(pairingheap *heap, pairingheap_node *node)
pairingheap * pairingheap_allocate(pairingheap_comparator compare, void *arg)
pairingheap_node * pairingheap_first(pairingheap *heap)
#define pairingheap_container(type, membername, ptr)
#define pairingheap_const_container(type, membername, ptr)
FormData_pg_attribute * Form_pg_attribute
#define qsort(a, b, c, d)
static Datum PointerGetDatum(const void *X)
static Oid DatumGetObjectId(Datum X)
static Pointer DatumGetPointer(Datum X)
static Datum Int32GetDatum(int32 X)
static int32 DatumGetInt32(Datum X)
static int fd(const char *x, int i)
MemoryContextSwitchTo(old_ctx)
#define RelationIsLogicallyLogged(relation)
#define RelationGetDescr(relation)
#define RelationGetRelationName(relation)
#define RelationIsValid(relation)
Relation RelationIdGetRelation(Oid relationId)
void RelationClose(Relation relation)
Oid RelidByRelfilenumber(Oid reltablespace, RelFileNumber relfilenumber)
#define relpathperm(rlocator, forknum)
static int file_sort_by_lsn(const ListCell *a_p, const ListCell *b_p)
void ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change, bool upd_mem)
static void ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
static void ReorderBufferStreamCommit(ReorderBuffer *rb, ReorderBufferTXN *txn)
void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)
static void ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
static ReorderBufferTXN * ReorderBufferLargestTXN(ReorderBuffer *rb)
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
static void ApplyLogicalMappingFile(HTAB *tuplecid_data, Oid relid, const char *fname)
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
static void ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn)
static void ReorderBufferToastInitHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, TimestampTz abort_time)
static bool ReorderBufferCanStartStreaming(ReorderBuffer *rb)
static void ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, Snapshot snapshot_now, CommandId command_id, XLogRecPtr last_lsn, ReorderBufferChange *specinsert)
struct ReorderBufferDiskChange ReorderBufferDiskChange
Oid * ReorderBufferGetRelids(ReorderBuffer *rb, int nrelids)
HeapTuple ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb)
static int ReorderBufferIterCompare(Datum a, Datum b, void *arg)
static void ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferIterTXNState *volatile *iter_state)
bool ResolveCminCmaxDuringDecoding(HTAB *tuplecid_data, Snapshot snapshot, HeapTuple htup, Buffer buffer, CommandId *cmin, CommandId *cmax)
static void ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
static void ReorderBufferReplay(ReorderBufferTXN *txn, ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, char *gid)
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
TransactionId * ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
static void ReorderBufferSaveTXNSnapshot(ReorderBuffer *rb, ReorderBufferTXN *txn, Snapshot snapshot_now, CommandId command_id)
#define IsSpecInsert(action)
static Size ReorderBufferChangeSize(ReorderBufferChange *change)
ReorderBuffer * ReorderBufferAllocate(void)
int logical_decoding_work_mem
static void AssertChangeLsnOrder(ReorderBufferTXN *txn)
static bool ReorderBufferCanStream(ReorderBuffer *rb)
static int ReorderBufferTXNSizeCompare(const pairingheap_node *a, const pairingheap_node *b, void *arg)
static void ReorderBufferApplyChange(ReorderBuffer *rb, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change, bool streaming)
void ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid)
static void ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn, int fd, ReorderBufferChange *change)
struct ReorderBufferIterTXNState ReorderBufferIterTXNState
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
struct ReorderBufferTXNByIdEnt ReorderBufferTXNByIdEnt
int debug_logical_replication_streaming
struct ReorderBufferIterTXNEntry ReorderBufferIterTXNEntry
static void ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
#define IsInsertOrUpdate(action)
static void ReorderBufferSerializeReserve(ReorderBuffer *rb, Size sz)
void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snap, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
static void ReorderBufferExecuteInvalidations(uint32 nmsgs, SharedInvalidationMessage *msgs)
static void ReorderBufferIterTXNFinish(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
struct RewriteMappingFile RewriteMappingFile
void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
static void ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, bool txn_prepared)
#define CHANGES_THRESHOLD
static ReorderBufferTXN * ReorderBufferLargestStreamableTopTXN(ReorderBuffer *rb)
static void ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn, char *data)
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)
static void AssertTXNLsnOrder(ReorderBuffer *rb)
static void ReorderBufferApplyMessage(ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferChange *change, bool streaming)
void ReorderBufferReturnTupleBuf(HeapTuple tuple)
static void ReorderBufferFreeSnap(ReorderBuffer *rb, Snapshot snap)
static void ReorderBufferCleanupSerializedTXNs(const char *slotname)
void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
static void SetupCheckXidLive(TransactionId xid)
static bool TransactionIdInArray(TransactionId xid, TransactionId *xip, Size num)
static Snapshot ReorderBufferCopySnap(ReorderBuffer *rb, Snapshot orig_snap, ReorderBufferTXN *txn, CommandId cid)
static void ReorderBufferApplyTruncate(ReorderBuffer *rb, ReorderBufferTXN *txn, int nrelations, Relation *relations, ReorderBufferChange *change, bool streaming)
static void ReorderBufferProcessPartialChange(ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferChange *change, bool toast_insert)
static void ReorderBufferToastReset(ReorderBuffer *rb, ReorderBufferTXN *txn)
static void ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
static void UpdateLogicalMappings(HTAB *tuplecid_data, Oid relid, Snapshot snapshot)
bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid, XLogRecPtr prepare_lsn, XLogRecPtr end_lsn, TimestampTz prepare_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
static void ReorderBufferTransferSnapToParent(ReorderBufferTXN *txn, ReorderBufferTXN *subtxn)
struct TXNEntryFile TXNEntryFile
static void ReorderBufferBuildTupleCidHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
void ReorderBufferReturnRelids(ReorderBuffer *rb, Oid *relids)
static ReorderBufferChange * ReorderBufferIterTXNNext(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
static void ReorderBufferCheckMemoryLimit(ReorderBuffer *rb)
static void ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb, ReorderBufferChange *change, ReorderBufferTXN *txn, bool addition, Size sz)
struct ReorderBufferToastEnt ReorderBufferToastEnt
static void ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn, TXNEntryFile *file, XLogSegNo *segno)
void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
void ReorderBufferFree(ReorderBuffer *rb)
static void ReorderBufferSerializedPath(char *path, ReplicationSlot *slot, TransactionId xid, XLogSegNo segno)
#define IsSpecConfirmOrAbort(action)
struct ReorderBufferTupleCidEnt ReorderBufferTupleCidEnt
struct ReorderBufferTupleCidKey ReorderBufferTupleCidKey
static const Size max_changes_in_memory
void StartupReorderBuffer(void)
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
static ReorderBufferTXN * ReorderBufferTXNByXid(ReorderBuffer *rb, TransactionId xid, bool create, bool *is_new, XLogRecPtr lsn, bool create_as_top)
ReorderBufferTXN * ReorderBufferGetOldestTXN(ReorderBuffer *rb)
static ReorderBufferTXN * ReorderBufferGetTXN(ReorderBuffer *rb)
static void ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr commit_lsn, volatile Snapshot snapshot_now, volatile CommandId command_id, bool streaming)
#define rbtxn_prepared(txn)
#define rbtxn_has_streamable_change(txn)
#define rbtxn_has_catalog_changes(txn)
@ DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE
@ DEBUG_LOGICAL_REP_STREAMING_BUFFERED
#define rbtxn_is_serialized_clear(txn)
#define RBTXN_IS_STREAMED
#define RBTXN_HAS_PARTIAL_CHANGE
#define rbtxn_is_streamed(txn)
struct ReorderBufferChange ReorderBufferChange
#define rbtxn_is_toptxn(txn)
#define rbtxn_get_toptxn(txn)
#define rbtxn_is_known_subxact(txn)
#define rbtxn_is_subtxn(txn)
#define RBTXN_HAS_CATALOG_CHANGES
#define PG_LOGICAL_MAPPINGS_DIR
#define RBTXN_IS_SERIALIZED_CLEAR
#define RBTXN_SKIPPED_PREPARE
#define RBTXN_HAS_STREAMABLE_CHANGE
@ REORDER_BUFFER_CHANGE_INVALIDATION
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM
@ REORDER_BUFFER_CHANGE_INSERT
@ REORDER_BUFFER_CHANGE_MESSAGE
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT
@ REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID
@ REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID
@ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT
@ REORDER_BUFFER_CHANGE_TRUNCATE
@ REORDER_BUFFER_CHANGE_DELETE
@ REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT
@ REORDER_BUFFER_CHANGE_UPDATE
#define RBTXN_IS_SERIALIZED
#define rbtxn_is_serialized(txn)
#define rbtxn_has_partial_change(txn)
#define LOGICAL_REWRITE_FORMAT
static pg_noinline void Size size
MemoryContext SlabContextCreate(MemoryContext parent, const char *name, Size blockSize, Size chunkSize)
ReplicationSlot * MyReplicationSlot
bool ReplicationSlotValidateName(const char *name, int elevel)
void SnapBuildSnapDecRefcount(Snapshot snap)
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
void TeardownHistoricSnapshot(bool is_error)
void SetupHistoricSnapshot(Snapshot historic_snapshot, HTAB *tuplecids)
static HTAB * tuplecid_data
struct SnapshotData * Snapshot
struct SnapshotData SnapshotData
struct SnapBuild * snapshot_builder
RelFileLocator old_locator
RelFileLocator new_locator
struct ReorderBufferChange::@109::@114 inval
ReorderBufferChangeType action
bool clear_toast_afterwards
struct ReorderBufferChange::@109::@111 truncate
struct ReorderBufferTXN * txn
struct ReorderBufferChange::@109::@113 tuplecid
struct ReorderBufferChange::@109::@110 tp
struct ReorderBufferChange::@109::@112 msg
SharedInvalidationMessage * invalidations
union ReorderBufferChange::@109 data
ReorderBufferChange change
ReorderBufferChange * change
ReorderBufferIterTXNEntry entries[FLEXIBLE_ARRAY_MEMBER]
XLogRecPtr restart_decoding_lsn
pairingheap_node txn_node
XLogRecPtr base_snapshot_lsn
TransactionId toplevel_xid
dlist_node catchange_node
SharedInvalidationMessage * invalidations
struct ReorderBufferTXN * toptxn
union ReorderBufferTXN::@115 xact_time
void * output_plugin_private
dlist_node base_snapshot_node
struct varlena * reconstructed
ReorderBufferTupleCidKey key
ReorderBufferStreamMessageCB stream_message
ReorderBufferStreamChangeCB stream_change
ReorderBufferBeginCB begin_prepare
ReorderBufferStreamTruncateCB stream_truncate
ReorderBufferCommitPreparedCB commit_prepared
ReorderBufferUpdateProgressTxnCB update_progress_txn
ReorderBufferMessageCB message
dlist_head txns_by_base_snapshot_lsn
dclist_head catchange_txns
ReorderBufferRollbackPreparedCB rollback_prepared
ReorderBufferPrepareCB prepare
ReorderBufferStreamStopCB stream_stop
ReorderBufferApplyChangeCB apply_change
MemoryContext change_context
ReorderBufferTXN * by_txn_last_txn
TransactionId by_txn_last_xid
ReorderBufferStreamPrepareCB stream_prepare
ReorderBufferStreamAbortCB stream_abort
MemoryContext tup_context
ReorderBufferCommitCB commit
ReorderBufferStreamStartCB stream_start
ReorderBufferStreamCommitCB stream_commit
ReorderBufferApplyTruncateCB apply_truncate
dlist_head toplevel_by_lsn
ReorderBufferBeginCB begin
MemoryContext txn_context
XLogRecPtr current_restart_decoding_lsn
ReplicationSlotPersistentData data
bool TransactionIdDidCommit(TransactionId transactionId)
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
#define InvalidTransactionId
#define TransactionIdEquals(id1, id2)
#define TransactionIdIsValid(xid)
static FormData_pg_attribute * TupleDescAttr(TupleDesc tupdesc, int i)
#define VARSIZE_SHORT(PTR)
#define VARATT_IS_EXTENDED(PTR)
#define VARATT_IS_SHORT(PTR)
#define SET_VARSIZE_COMPRESSED(PTR, len)
#define SET_VARTAG_EXTERNAL(PTR, tag)
#define VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer)
#define VARDATA_EXTERNAL(PTR)
#define SET_VARSIZE(PTR, len)
#define VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer)
#define VARATT_IS_EXTERNAL(PTR)
static void pgstat_report_wait_start(uint32 wait_event_info)
static void pgstat_report_wait_end(void)
bool IsTransactionOrTransactionBlock(void)
void BeginInternalSubTransaction(const char *name)
TransactionId CheckXidAlive
void RollbackAndReleaseCurrentSubTransaction(void)
void StartTransactionCommand(void)
TransactionId GetCurrentTransactionIdIfAny(void)
TransactionId GetCurrentTransactionId(void)
void AbortCurrentTransaction(void)
int xidComparator(const void *arg1, const void *arg2)
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define XLByteInSeg(xlrp, logSegNo, wal_segsz_bytes)
#define LSN_FORMAT_ARGS(lsn)
#define InvalidXLogRecPtr