181 #define IsSpecInsert(action) \
183 ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT) \
185 #define IsSpecConfirmOrAbort(action) \
187 (((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM) || \
188 ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT)) \
190 #define IsInsertOrUpdate(action) \
192 (((action) == REORDER_BUFFER_CHANGE_INSERT) || \
193 ((action) == REORDER_BUFFER_CHANGE_UPDATE) || \
194 ((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT)) \
297 bool addition,
Size sz);
320 memset(&hash_ctl, 0,
sizeof(hash_ctl));
441 if (txn->
gid != NULL)
499 if (change->
data.
tp.newtuple)
502 change->
data.
tp.newtuple = NULL;
505 if (change->
data.
tp.oldtuple)
508 change->
data.
tp.oldtuple = NULL;
512 if (change->
data.
msg.prefix != NULL)
514 change->
data.
msg.prefix = NULL;
515 if (change->
data.
msg.message != NULL)
517 change->
data.
msg.message = NULL;
564 MAXIMUM_ALIGNOF + alloc_len);
595 alloc_len =
sizeof(
Oid) * nrelids;
619 bool *is_new,
XLogRecPtr lsn,
bool create_as_top)
693 Assert(!create || txn != NULL);
740 change->
data.
tp.clear_toast_afterwards)
840 bool transactional,
const char *prefix,
841 Size message_size,
const char *message)
862 change->
data.
msg.message_size = message_size;
864 memcpy(change->
data.
msg.message, message, message_size);
873 volatile Snapshot snapshot_now = snap;
885 rb->
message(rb, txn, lsn,
false, prefix, message_size, message);
909 #ifdef USE_ASSERT_CHECKING
942 Assert(prev_first_lsn < cur_txn->first_lsn);
962 Assert(prev_base_snap_lsn < cur_txn->base_snapshot_lsn);
980 #ifdef USE_ASSERT_CHECKING
997 Assert(prev_lsn <= cur_change->lsn);
999 prev_lsn = cur_change->
lsn;
1234 else if (pos_a == pos_b)
1289 state->nr_txns = nr_txns;
1292 for (off = 0; off <
state->nr_txns; off++)
1294 state->entries[off].file.vfd = -1;
1295 state->entries[off].segno = 0;
1304 *iter_state =
state;
1323 &
state->entries[off].segno);
1329 state->entries[off].lsn = cur_change->
lsn;
1330 state->entries[off].change = cur_change;
1331 state->entries[off].txn = txn;
1352 &
state->entries[off].file,
1353 &
state->entries[off].segno);
1358 state->entries[off].lsn = cur_change->
lsn;
1359 state->entries[off].change = cur_change;
1360 state->entries[off].txn = cur_txn;
1384 if (
state->heap->bh_size == 0)
1388 entry = &
state->entries[off];
1414 state->entries[off].lsn = next_change->
lsn;
1415 state->entries[off].change = next_change;
1438 &
state->entries[off].segno))
1445 elog(
DEBUG2,
"restored %u/%u changes from disk",
1451 state->entries[off].lsn = next_change->
lsn;
1452 state->entries[off].change = next_change;
1474 for (off = 0; off <
state->nr_txns; off++)
1476 if (
state->entries[off].file.vfd != -1)
1972 int nrelations,
Relation *relations,
1991 change->
data.
msg.message_size,
1996 change->
data.
msg.message_size,
2011 if (snapshot_now->
copied)
2039 if (specinsert != NULL)
2080 volatile bool stream_started =
false;
2104 int changes_count = 0;
2142 stream_started =
true;
2153 prev_lsn = change->
lsn;
2162 curtxn = change->
txn;
2175 if (specinsert == NULL)
2176 elog(
ERROR,
"invalid ordering of speculative insertion changes");
2178 change = specinsert;
2188 change->
data.
tp.rlocator.relNumber);
2203 change->
data.
tp.newtuple == NULL &&
2204 change->
data.
tp.oldtuple == NULL)
2207 elog(
ERROR,
"could not map filenumber \"%s\" to relation OID",
2214 elog(
ERROR,
"could not open relation with OID %u (for filenumber \"%s\")",
2235 if (relation->
rd_rel->relkind == RELKIND_SEQUENCE)
2250 if (change->
data.
tp.clear_toast_afterwards)
2277 if (specinsert != NULL)
2306 if (specinsert != NULL)
2314 specinsert = change;
2327 if (specinsert != NULL)
2351 for (
i = 0;
i < nrelids;
i++)
2359 elog(
ERROR,
"could not open relation with OID %u", relid);
2364 relations[nrelations++] = rel;
2372 for (
i = 0;
i < nrelations;
i++)
2392 if (snapshot_now->
copied)
2423 if (command_id < change->
data.command_id)
2427 if (!snapshot_now->
copied)
2434 snapshot_now->
curcid = command_id;
2443 elog(
ERROR,
"tuplecid value in changequeue");
2457 #define CHANGES_THRESHOLD 100
2496 stream_started =
false;
2506 rb->
prepare(rb, txn, commit_lsn);
2508 rb->
commit(rb, txn, commit_lsn);
2513 elog(
ERROR,
"output plugin used XID %u",
2522 else if (snapshot_now->
copied)
2601 if (errdata->
sqlerrcode == ERRCODE_TRANSACTION_ROLLBACK &&
2615 command_id, prev_lsn,
2715 origin_id, origin_lsn);
2810 XLogRecPtr origin_lsn,
char *gid,
bool is_commit)
2826 prepare_end_lsn = txn->
end_lsn;
2840 if ((txn->
final_lsn < two_phase_at) && is_commit)
3073 for (
i = 0;
i < ninvalidations;
i++)
3180 bool addition,
Size sz)
3312 change->
data.
inval.ninvalidations = nmsgs;
3315 memcpy(change->
data.
inval.invalidations, msgs,
3332 for (
i = 0;
i < nmsgs;
i++)
3399 xids[xcnt++] = txn->
xid;
3498 if ((!largest) || (txn->
size > largest->
size))
3537 Size largest_size = 0;
3552 if ((largest == NULL || txn->
total_size > largest_size) &&
3659 elog(
DEBUG2,
"spill %u changes in XID %u to disk",
3701 O_CREAT | O_WRONLY | O_APPEND |
PG_BINARY);
3706 errmsg(
"could not open file \"%s\": %m", path)));
3767 oldtup = change->
data.
tp.oldtuple;
3768 newtup = change->
data.
tp.newtuple;
3813 Size prefix_size = strlen(change->
data.
msg.prefix) + 1;
3815 sz += prefix_size + change->
data.
msg.message_size +
3825 memcpy(
data, &prefix_size,
sizeof(
Size));
3829 data += prefix_size;
3835 change->
data.
msg.message_size);
3928 int save_errno = errno;
3933 errno = save_errno ? save_errno : ENOSPC;
3936 errmsg(
"could not write to data file for XID %u: %m",
3997 bool txn_is_streamed;
4126 oldtup = change->
data.
tp.oldtuple;
4127 newtup = change->
data.
tp.newtuple;
4147 Size prefix_size = strlen(change->
data.
msg.prefix) + 1;
4149 sz += prefix_size + change->
data.
msg.message_size +
4248 if (*
fd < 0 && errno == ENOENT)
4257 errmsg(
"could not open file \"%s\": %m",
4269 file->
curOffset, WAIT_EVENT_REORDER_BUFFER_READ);
4279 else if (readBytes < 0)
4282 errmsg(
"could not read from reorderbuffer spill file: %m")));
4286 errmsg(
"could not read from reorderbuffer spill file: read %d instead of %u bytes",
4302 WAIT_EVENT_REORDER_BUFFER_READ);
4307 errmsg(
"could not read from reorderbuffer spill file: %m")));
4311 errmsg(
"could not read from reorderbuffer spill file: read %d instead of %u bytes",
4360 if (change->
data.
tp.oldtuple)
4364 change->
data.
tp.oldtuple =
4368 memcpy(&change->
data.
tp.oldtuple->tuple,
data,
4373 change->
data.
tp.oldtuple->tuple.t_data =
4377 memcpy(change->
data.
tp.oldtuple->tuple.t_data,
data, tuplelen);
4381 if (change->
data.
tp.newtuple)
4389 change->
data.
tp.newtuple =
4393 memcpy(&change->
data.
tp.newtuple->tuple,
data,
4398 change->
data.
tp.newtuple->tuple.t_data =
4402 memcpy(change->
data.
tp.newtuple->tuple.t_data,
data, tuplelen);
4412 memcpy(&prefix_size,
data,
sizeof(
Size));
4416 memcpy(change->
data.
msg.prefix,
data, prefix_size);
4418 data += prefix_size;
4424 change->
data.
msg.message_size);
4426 change->
data.
msg.message_size);
4460 memcpy(newsnap,
data, size);
4523 if (unlink(path) != 0 && errno != ENOENT)
4526 errmsg(
"could not remove file \"%s\": %m", path)));
4539 struct stat statbuf;
4542 sprintf(path,
"pg_replslot/%s", slotname);
4552 if (strncmp(spill_de->
d_name,
"xid", 3) == 0)
4555 "pg_replslot/%s/%s", slotname,
4558 if (unlink(path) != 0)
4561 errmsg(
"could not remove file \"%s\" during removal of pg_replslot/%s/xid*: %m",
4594 struct dirent *logical_de;
4597 while ((logical_de =
ReadDir(logical_dir,
"pg_replslot")) != NULL)
4599 if (strcmp(logical_de->
d_name,
".") == 0 ||
4600 strcmp(logical_de->
d_name,
"..") == 0)
4663 newtup = change->
data.
tp.newtuple;
4682 elog(
ERROR,
"got sequence entry %d for toast chunk %u instead of seq 0",
4683 chunk_seq, chunk_id);
4686 elog(
ERROR,
"got sequence entry %d for toast chunk %u instead of seq %d",
4699 elog(
ERROR,
"unexpected type of toast chunk");
4701 ent->
size += chunksize;
4768 elog(
ERROR,
"could not open toast relation with OID %u (base relation \"%s\")",
4778 newtup = change->
data.
tp.newtuple;
4782 for (natt = 0; natt < desc->
natts; natt++)
4791 struct varlena *new_datum = NULL;
4792 struct varlena *reconstructed;
4797 if (attr->attnum < 0)
4800 if (attr->attisdropped)
4804 if (attr->attlen != -1)
4849 ctup = cchange->
data.
tp.newtuple;
4856 memcpy(
VARDATA(reconstructed) + data_done,
4869 memset(&redirect_pointer, 0,
sizeof(redirect_pointer));
4870 redirect_pointer.
pointer = reconstructed;
4874 sizeof(redirect_pointer));
4897 for (natt = 0; natt < desc->
natts; natt++)
4998 elog(
DEBUG3,
"mapping: node: %u/%u/%u tid: %u/%u cmin: %u, cmax: %u",
5025 sprintf(path,
"pg_logical/mappings/%s", fname);
5030 errmsg(
"could not open file \"%s\": %m", path)));
5050 errmsg(
"could not read file \"%s\": %m",
5052 else if (readBytes == 0)
5057 errmsg(
"could not read from file \"%s\": read %d instead of %d bytes",
5102 errmsg(
"could not close file \"%s\": %m", path)));
5112 return bsearch(&xid, xip, num,
5125 if (
a->lsn <
b->lsn)
5127 else if (
a->lsn >
b->lsn)
5140 struct dirent *mapping_de;
5146 while ((mapping_de =
ReadDir(mapping_dir,
"pg_logical/mappings")) != NULL)
5157 if (strcmp(mapping_de->
d_name,
".") == 0 ||
5158 strcmp(mapping_de->
d_name,
"..") == 0)
5162 if (strncmp(mapping_de->
d_name,
"map-", 4) != 0)
5166 &f_dboid, &f_relid, &f_hi, &f_lo,
5167 &f_mapped_xid, &f_create_xid) != 6)
5170 f_lsn = ((uint64) f_hi) << 32 | f_lo;
5173 if (f_dboid != dboid)
5177 if (f_relid != relid)
5199 foreach(file, files)
5224 bool updated_mapping =
false;
5238 memset(&
key, 0,
sizeof(
key));
5265 if (ent == NULL && !updated_mapping)
5269 updated_mapping =
true;
5272 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)
binaryheap * binaryheap_allocate(int capacity, binaryheap_comparator compare, void *arg)
void binaryheap_free(binaryheap *heap)
void binaryheap_add_unordered(binaryheap *heap, bh_node_type d)
static void cleanup(void)
#define BufferIsLocal(buffer)
void BufferGetTag(Buffer buffer, RelFileLocator *rlocator, ForkNumber *forknum, BlockNumber *blknum)
#define FLEXIBLE_ARRAY_MEMBER
bool IsToastRelation(Relation relation)
bool IsSharedRelation(Oid relationId)
elog(ERROR, "%s: %s", p2, msg)
#define INDIRECT_POINTER_SIZE
#define VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr)
void hash_destroy(HTAB *hashp)
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
void * hash_seq_search(HASH_SEQ_STATUS *status)
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
void FreeErrorData(ErrorData *edata)
int errcode_for_file_access(void)
void FlushErrorState(void)
int errmsg(const char *fmt,...)
ErrorData * CopyErrorData(void)
#define ereport(elevel,...)
struct dirent * ReadDir(DIR *dir, const char *dirname)
int CloseTransientFile(int fd)
void FileClose(File file)
File PathNameOpenFile(const char *fileName, int fileFlags)
int FileRead(File file, void *buffer, size_t amount, off_t offset, uint32 wait_event_info)
struct dirent * ReadDirExtended(DIR *dir, const char *dirname, int elevel)
int OpenTransientFile(const char *fileName, int fileFlags)
DIR * AllocateDir(const char *dirname)
MemoryContext GenerationContextCreate(MemoryContext parent, const char *name, Size minContextSize, Size initBlockSize, Size maxBlockSize)
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
void heap_deform_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *values, bool *isnull)
HeapTupleData * HeapTuple
struct HeapTupleData HeapTupleData
#define SizeofHeapTupleHeader
static Datum fastgetattr(HeapTuple tup, int attnum, TupleDesc tupleDesc, bool *isnull)
#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 void dlist_delete(dlist_node *node)
static uint32 dclist_count(const dclist_head *head)
static dlist_node * dlist_next_node(dlist_head *head, dlist_node *node)
static dlist_node * dlist_pop_head_node(dlist_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)
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)
Assert(fmt[strlen(fmt) - 1] !='\n')
void list_sort(List *list, list_sort_comparator cmp)
List * lappend(List *list, void *datum)
void UpdateDecodingStats(LogicalDecodingContext *ctx)
char * pstrdup(const char *in)
void pfree(void *pointer)
void * palloc0(Size size)
void * MemoryContextAllocZero(MemoryContext context, Size size)
MemoryContext CurrentMemoryContext
void * repalloc(void *pointer, Size size)
void * MemoryContextAlloc(MemoryContext context, Size size)
void MemoryContextDelete(MemoryContext context)
#define AllocSetContextCreate
#define ALLOCSET_DEFAULT_SIZES
#define SLAB_DEFAULT_BLOCK_SIZE
#define SLAB_LARGE_BLOCK_SIZE
#define CHECK_FOR_INTERRUPTS()
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
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)
#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)
TransactionId * ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
void ReorderBufferReturnTupleBuf(ReorderBuffer *rb, ReorderBufferTupleBuf *tuple)
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)
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, XLog