huawei
/
openGauss-server
0
0
Fork 77
Code Issues Pull Requests 24 Projects Releases Wiki Activity

openGauss代码评注赛——ustore堆表和索引相关注释代码 #41

Open
njuzhouzc wants to merge 4 commits from njuzhouzc/openGauss-server:3.0.0 into 3.0.0
pull from: njuzhouzc/openGauss-server:3.0.0
merge into: huawei:3.0.0
Conversation 0 Commits 4 Files Changed 3 +260 -13
3 changed files with 260 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

164
src/gausskernel/storage/access/ubtree/ubtrecycle.cpp
View File

@ -79,6 +79,18 @@ static UBTRecycleQueueItem HeaderGetItem(UBTRecycleQueueHeader header, uint16 of
pg_unreachable(); /* won't reach here */
}
/*
* UBTreeInitRecycleQueuePage
*
* Initialize a page of the recycle queue
*
* This function will be called when the system is initializing the recycle queue,
* or extend one page of the recycle queue.
* The function will initialize the page usiong the starting value, if it is a meta
* page, the functionm will initialize the meta data.
*
* NOTE: the pages in URQ doesn't have special space.
*/
static void UBTreeInitRecycleQueuePage(Relation rel, Page page, Size size, BlockNumber blkno)
{
PageInit(page, size, 0);
@ -113,6 +125,7 @@ static void UBTreeInitRecycleQueuePage(Relation rel, Page page, Size size, Block
}
}
/* check whether the URQ has been initialized */
static bool RecycleQueueInitialized(Relation rel)
{
/* open smgr, might have to re-open if a cache flush happened */
@ -219,6 +232,15 @@ static void InitRecycleQueueInitialPage(Relation rel, Buffer buf)
END_CRIT_SECTION();
}
/*
* ReadRecycleQueueBuffer
*
* Read one page of the recycle queue and return it.
*
* We use forknum FSM_FORKNUM to acquire the page of the recycle queue. Note that
* the files of this forknumber do not actually store the structure of fsm, they
* store URQ instead.
*/
Buffer ReadRecycleQueueBuffer(Relation rel, BlockNumber blkno)
{
Buffer buf = ReadBufferExtended(rel, FSM_FORKNUM, blkno, RBM_NORMAL, NULL);
@ -231,6 +253,19 @@ Buffer ReadRecycleQueueBuffer(Relation rel, BlockNumber blkno)
return buf;
}
/*
* UBTreeInitializeRecycleQueue
*
* Initialize a recycle queue for a certain UBtree relation
*
* This is the entrance function of the URQ, and will be called during UBTreeLoad,
* which is a subroutine of constructing a UBtree index. The function will also be
* called if we need to use URQ but it hasn't been initialized.
* This function will create two recycle queues and initialize them, one is Potential
* Empty Page Queue which has pages 0 2 4, and the other is Available Page Queue which
* has pages 1 3 5. First, check existing pages, and if the pages are not enough, then
* create necessary pages(URQ has at least 6 pages).
*/
void UBTreeInitializeRecycleQueue(Relation rel)
{
LockRelationForExtension(rel, ExclusiveLock);
@ -306,6 +341,7 @@ static bool UBTreeTryRecycleEmptyPageInternal(Relation rel)
return true;
}
/* Try to move pages from Potential Empty Page Queue to Available Page Queue */
void UBTreeTryRecycleEmptyPage(Relation rel)
{
bool firstTrySucceed = UBTreeTryRecycleEmptyPageInternal(rel);
@ -315,16 +351,34 @@ void UBTreeTryRecycleEmptyPage(Relation rel)
}
}
/*
* UBTreeRecordFreePage
*
* Put a freed page into the Available Page Queue
*
* This function will be called if the page has been deleted and moved out of
* the UBtree, by the function UBTreeVacuumPage. We can safely put it in URQ
* for reuse.
*/
void UBTreeRecordFreePage(Relation rel, BlockNumber blkno, TransactionId xid)
{
UBTreeRecycleQueueAddPage(rel, RECYCLE_FREED_FORK, blkno, xid);
}
/*
* UBTreeRecordEmptyPage
*
* Put a (possibly) empty page into the Potential Empty Page Queue
*
* This function will be called if the page's opaque->activeTupleCount = 0, as
* a subroutine of UBTreeDoDelete.
*/
void UBTreeRecordEmptyPage(Relation rel, BlockNumber blkno, TransactionId xid)
{
UBTreeRecycleQueueAddPage(rel, RECYCLE_EMPTY_FORK, blkno, xid);
}
/* When we reuse a UBtree page from URQ, we remove the item which matches the page. */
void UBTreeRecordUsedPage(Relation rel, UBTRecycleQueueAddress addr)
{
if (addr.queueBuf != InvalidBuffer) {
@ -332,6 +386,7 @@ void UBTreeRecordUsedPage(Relation rel, UBTRecycleQueueAddress addr)
}
}
/* Step next page behind current page buf */
static Buffer StepNextPage(Relation rel, Buffer buf)
{
Page page = BufferGetPage(buf);
@ -346,6 +401,17 @@ static Buffer StepNextPage(Relation rel, Buffer buf)
return nextBuf;
}
/*
* GetAvailablePageOnPage
*
* Get an available UBTree page info from the URQ page.
*
* Here, the function name means: get an available "UBtree Page" information from
* a "URQ Page".
* We get the header of the page and check page info items. If we find a proper
* UBtree page, we collect its info in the parameter addr and return it. Else, we
* return InvalidBuffer.
*/
static Buffer GetAvailablePageOnPage(Relation rel, UBTRecycleForkNumber forkNumber, Buffer buf,
TransactionId WaterLevelXid, UBTRecycleQueueAddress *addr, bool *continueScan)
{
@ -356,6 +422,12 @@ static Buffer GetAvailablePageOnPage(Relation rel, UBTRecycleForkNumber forkNumb
while (IsNormalOffset(curOffset)) {
UBTRecycleQueueItem item = HeaderGetItem(header, curOffset);
if (TransactionIdFollowsOrEquals(item->xid, WaterLevelXid)) {
/*
* We should know that the item->xid in the URQ are increasing, so the xid
* of the latter items are greater than or equal to item->xid here, and
* also the WaterLevelXid. So we needn't to scan the latter items in the
* URQ, and just return InvalidBuffer.
*/
*continueScan = false;
return InvalidBuffer;
}
@ -370,6 +442,7 @@ static Buffer GetAvailablePageOnPage(Relation rel, UBTRecycleForkNumber forkNumb
if (forkNumber == RECYCLE_EMPTY_FORK || UBTreePageRecyclable(BufferGetPage(targetBuf))) {
WHITEBOX_TEST_STUB("GetAvailablePageOnPage-got", WhiteboxDefaultErrorEmit);
// Here we get an available UBtree page info, so we store it and return
*continueScan = false;
addr->queueBuf = buf;
addr->indexBlkno = item->blkno;
@ -392,6 +465,18 @@ static Buffer GetAvailablePageOnPage(Relation rel, UBTRecycleForkNumber forkNumb
return InvalidBuffer;
}
/*
* UBTreeGetAvailablePage
*
* Get an available page from the URQ(Always Available Page Queue)
*
* This is the function which we use to acquire used pages from URQ for recycle.
* 1.We first get the head page of the URQ, and then get an available page from it,
* if we couldn't get an available page, we step to next queue page.
* 2.If we get an available page, we can return the page. If we couldn't get one and
* the forknumber is EMPTY, we just return InvalidBuffer. Else, do step 3.
* 3.Here, forknumber is FREED, we check newly created pages in MAIN_FORKNUM.
*/
Buffer UBTreeGetAvailablePage(Relation rel, UBTRecycleForkNumber forkNumber, UBTRecycleQueueAddress *addr)
{
TransactionId frozenXid = g_instance.undo_cxt.oldestFrozenXid;
@ -399,6 +484,7 @@ Buffer UBTreeGetAvailablePage(Relation rel, UBTRecycleForkNumber forkNumber, UBT
TransactionId waterLevelXid = ((forkNumber == RECYCLE_EMPTY_FORK) ? recycleXid : frozenXid);
// Get the head page of the URQ.
Buffer queueBuf = RecycleQueueGetEndpointPage(rel, forkNumber, true, BT_READ);
Buffer indexBuf = InvalidBuffer;
@ -432,6 +518,7 @@ Buffer UBTreeGetAvailablePage(Relation rel, UBTRecycleForkNumber forkNumber, UBT
Buffer metaBuf = ReadRecycleQueueBuffer(rel, metaBlockNumber);
LockBuffer(metaBuf, BT_READ);
UBTRecycleMeta metaData = (UBTRecycleMeta)PageGetContents(BufferGetPage(metaBuf));
// NOTE: At the time we inited the URQ, we executed meta->nblocksUpper = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM);
for (BlockNumber curBlkno = metaData->nblocksUpper; curBlkno < nblocks; curBlkno++) {
if (metaData->nblocksUpper > curBlkno) {
continue;
@ -460,14 +547,25 @@ Buffer UBTreeGetAvailablePage(Relation rel, UBTRecycleForkNumber forkNumber, UBT
return indexBuf;
}
/*
* UBTreeRecycleQueuePageChangeEndpointLeftPage
*
* 1. isHead = true, it means that buf is the old head page, so remove
* its flag and set its head and tail
* 2. isHead = false, it means that buf is the old tail page, so remove
* its flag and set tailItem->next to indicate the next item is on
* the next new page.
*/
void UBTreeRecycleQueuePageChangeEndpointLeftPage(Buffer buf, bool isHead)
{
uint32 endpointFlag = (isHead ? URQ_HEAD_PAGE : URQ_TAIL_PAGE);
UBTRecycleQueueHeader header = GetRecycleQueueHeader(BufferGetPage(buf), BufferGetBlockNumber(buf));
if (isHead) {
/* buf is old head page */
header->head = InvalidOffset;
header->tail = InvalidOffset;
} else {
/* buf is old tail page */
Assert(IsNormalOffset(header->tail));
UBTRecycleQueueItem tailItem = HeaderGetItem(header, header->tail);
tailItem->next = OtherBlockOffset;
@ -475,11 +573,20 @@ void UBTreeRecycleQueuePageChangeEndpointLeftPage(Buffer buf, bool isHead)
header->flags &= ~endpointFlag;
}
/*
* UBTreeRecycleQueuePageChangeEndpointRightPage
*
* 1. isHead = true, it means that buf is the new head page, so set its
* flag and set its headItem->prev or header->head
* 2. isHead = false, it means that buf is the new tail page, so set its
* flag and check whether it has no item.
*/
void UBTreeRecycleQueuePageChangeEndpointRightPage(Buffer buf, bool isHead)
{
uint32 endpointFlag = (isHead ? URQ_HEAD_PAGE : URQ_TAIL_PAGE);
UBTRecycleQueueHeader header = GetRecycleQueueHeader(BufferGetPage(buf), BufferGetBlockNumber(buf));
if (isHead) {
/* buf is new head page */
if (IsNormalOffset(header->head)) {
UBTRecycleQueueItem headItem = HeaderGetItem(header, header->head);
headItem->prev = InvalidOffset;
@ -487,12 +594,13 @@ void UBTreeRecycleQueuePageChangeEndpointRightPage(Buffer buf, bool isHead)
header->head = InvalidOffset;
}
} else {
/* new created tail page must be empty */
/* buf is new tail page, newly created tail page must be empty */
Assert(header->head == InvalidOffset);
}
header->flags |= endpointFlag;
}
/* Change end point(head or tail) from buf to newBuf */
static void RecycleQueueChangeEndpoint(Relation rel, Buffer buf, Buffer nextBuf, bool isHead)
{
Page page = BufferGetPage(buf);
@ -530,6 +638,14 @@ static void RecycleQueueChangeEndpoint(Relation rel, Buffer buf, Buffer nextBuf,
END_CRIT_SECTION();
}
/*
* MoveToEndpointPage
*
* Move to the head or tail page of the URQ
* needHead ? head page : tail page
* If we find the head page with headItem->blkno = InvalidBlockNumber, we
* need to switch head.
*/
static Buffer MoveToEndpointPage(Relation rel, Buffer buf, bool needHead, int access)
{
restart:
@ -537,6 +653,7 @@ restart:
UBTRecycleQueueHeader header = GetRecycleQueueHeader(page, BufferGetBlockNumber(buf));
uint32 endpointFlag = (needHead ? URQ_HEAD_PAGE : URQ_TAIL_PAGE);
/* Start from the buf that we got from meta page */
while ((header->flags & endpointFlag) == 0) {
buf = StepNextPage(rel, buf);
page = BufferGetPage(buf);
@ -580,6 +697,13 @@ restart:
return buf;
}
/*
* PageAllocateItem
*
* Allocate an item offset for a certain UBtree page info
* URQ page will firstly try to allocate offset from freeItems, if not success,
* try to allocate offset from freeListHead.
*/
static uint16 PageAllocateItem(Buffer buf)
{
Page page = BufferGetPage(buf);
@ -604,6 +728,11 @@ static Buffer RecycleQueueExtend(Relation rel)
return buf;
}
/*
* RecycleQueueLinkNewPage
*
* Link a new URQ page at the end of the queue
*/
static void RecycleQueueLinkNewPage(Relation rel, Buffer leftBuf, Buffer newBuf)
{
/* new page already allocated, link it into the list */
@ -648,6 +777,7 @@ static void RecycleQueueLinkNewPage(Relation rel, Buffer leftBuf, Buffer newBuf)
UnlockReleaseBuffer(rightBuf);
}
/* Check whether a certain queue page is empty */
static bool QueuePageIsEmpty(Buffer buf)
{
UBTRecycleQueueHeader header = GetRecycleQueueHeader(BufferGetPage(buf), BufferGetBlockNumber(buf));
@ -706,6 +836,14 @@ static void TryFixMetaData(Buffer metaBuf, int32 oldval, int32 newval, bool isHe
}
}
/*
* RecycleQueueGetEndpointPage
*
* Get the head or the tail page of the URQ
*
* needHead ? head page : tail page
* If the end page has been changed during this function, we need to fix meta data.
*/
Buffer RecycleQueueGetEndpointPage(Relation rel, UBTRecycleForkNumber forkNumber, bool needHead, int access)
{
if (!RecycleQueueInitialized(rel)) {
@ -772,6 +910,13 @@ static void LogModifyPage(Buffer buf, bool isInsert, uint16 offset, UBTRecycleQu
PageSetLSN(page, recptr);
}
/*
* InsertOnRecycleQueuePage
*
* Insert a UBtree page info into the URQ page
* The item->xid in the URQ page is gradually increasing, so we need to find
* the proper place to insert our item.
*/
static void InsertOnRecycleQueuePage(Relation rel, Buffer buf, uint16 offset, BlockNumber blkno, TransactionId xid)
{
Page page = BufferGetPage(buf);
@ -880,6 +1025,15 @@ void UBTreeXlogRecycleQueueModifyPage(Buffer buf, xl_ubtree2_recycle_queue_modif
}
}
/*
* RemoveOneItemFromPage
*
* Remove the UBtree page info item from the queue page
* If the item is headitem or tail item of the queue page, just delete it.
* Else, we acquire the prev and next item of it and connect them. Then,
* clear the item, and insert it to freeListHead using the head insertion
* method. If we delete the last item of this page, we need to change head.
*/
static void RemoveOneItemFromPage(Relation rel, Buffer buf, uint16 offset)
{
Page page = BufferGetPage(buf);
@ -936,6 +1090,14 @@ static void RemoveOneItemFromPage(Relation rel, Buffer buf, uint16 offset)
}
}
/*
* UBTreeRecycleQueueDiscardPage
*
* Discard a certain UBtree page info from URQ
* The parameter addr pointers out the place where the info stores. We get
* the queue page and remove the item. And by the way, we try to remove the
* the invalid items in the left and right side of the item.
*/
static void UBTreeRecycleQueueDiscardPage(Relation rel, UBTRecycleQueueAddress addr)
{
Buffer buf = addr.queueBuf;

102
src/gausskernel/storage/access/ustore/knl_uheap.cpp
View File

@ -489,6 +489,19 @@ static ShortTransactionId UHeapTupleSetModifiedXid(Relation relation,
return tupleXid;
}
/*
* UHeapInsert - insert a heap tuple in the table
*
* Params:
* @param[IN] rel: the uheap relation
* @param[IN] utuple: the uheap tuple that will be inserted
* @param[IN] cid: CID of the inserting transaction
* @param[IN] bistate: state for bulk inserts.
* @param[IN] isToast: is toast or not.
*
* For most cases, the returned value is InvalidOid, for we successfully insert a tuple
* into the relation table.
*/
Oid UHeapInsert(RelationData *rel, UHeapTupleData *utuple, CommandId cid, BulkInsertState bistate, bool isToast)
{
Page page;
@ -511,9 +524,13 @@ Oid UHeapInsert(RelationData *rel, UHeapTupleData *utuple, CommandId cid, BulkIn
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("The insert tuple is NULL")));
}
Assert(utuple->tupTableType == UHEAP_TUPLE);
TransactionId fxid = GetTopTransactionId();
TransactionId fxid = GetTopTransactionId(); // get the xid of the main transaction
/* Prepare the tuple for insertion */
/*
* Prepare the tuple for insertion
* Here we should notice that "tuple" and "utuple" point to the same place in memory.
* We fill in some basic information based on the rel into the tuple.
*/
tuple = UHeapPrepareInsert(rel, utuple, 0);
/* Prepare Undo record before buffer lock since undo record length is fixed */
@ -603,6 +620,7 @@ reacquire_buffer:
/* Put utuple into buffer page */
RelationPutUTuple(rel, buffer, tuple);
// update potential free space of the page
UHeapRecordPotentialFreeSpace(buffer, -1 * SHORTALIGN(tuple->disk_tuple_size));
/* Update the UndoRecord now that we know where the tuple is located on the Page */
@ -747,16 +765,18 @@ TransactionId UHeapFetchInsertXid(UHeapTuple uhtup, Buffer buffer)
return result;
}
/* Put utuple into a page */
void RelationPutUTuple(Relation relation, Buffer buffer, UHeapTupleData *tuple)
{
OffsetNumber offNum = InvalidOffsetNumber;
UHeapBufferPage bufpage = {buffer, NULL};
// UPageAddItem: put the tuple into the buffer page
offNum =
UPageAddItem(relation, &bufpage, (Item)tuple->disk_tuple, tuple->disk_tuple_size, InvalidOffsetNumber, false);
if (offNum == InvalidOffsetNumber)
elog(PANIC, "failed to add tuple to page");
// set item pointer of the tuple->ctid
ItemPointerSet(&(tuple->ctid), BufferGetBlockNumber(buffer), offNum);
}
@ -767,7 +787,7 @@ UHeapTuple UHeapPrepareInsert(Relation rel, UHeapTupleData *tuple, int options)
tuple->disk_tuple->flag &= ~UHEAP_VIS_STATUS_MASK;
tuple->disk_tuple->td_id = UHEAPTUP_SLOT_FROZEN;
tuple->disk_tuple->locker_td_id = UHEAPTUP_SLOT_FROZEN;
tuple->table_oid = RelationGetRelid(rel);
tuple->table_oid = RelationGetRelid(rel); // Here, get the relid of the relation
tuple->t_bucketId = InvalidBktId;
if (rel->rd_rel->relkind != RELKIND_RELATION) {
@ -1672,6 +1692,24 @@ bool TableFetchAndStore(Relation scanRelation, Snapshot snapshot, Tuple tuple, B
return false;
}
/*
* UHeapDelete - delete a heap tuple in the table
*
* Params:
* @param[IN] relation: the uheap relation
* @param[IN] tid: the position of the uheap tuple in the table
* @param[IN] cid: CID of the deleting transaction
* @param[IN] crosscheck: snapshot to use for checking old tuple's visibility.
* @param[IN] snapshot: snapshot of the current transaction.
* @param[IN] wait: if true, wait for conflicting transactions to end.
* @param[OUT] oldslot: slot to store the old tuple in.
* @param[OUT] tmfd: filled with info about the old tuple.
* @param[IN] changingPart: true if we need to move into another partition.
* @param[IN] allowDeleteSelf: use in UHeapTupleSatisfiesUpdate for checking.
*
* For most cases, the returned value is TM_Ok, for we successfully delete a tuple
* from the relation. The old tuple will be placed into te undo zone.
*/
TM_Result UHeapDelete(Relation relation, ItemPointer tid, CommandId cid, Snapshot crosscheck, Snapshot snapshot,
bool wait, TupleTableSlot** oldslot, TM_FailureData *tmfd, bool changingPart, bool allowDeleteSelf)
{
@ -1711,10 +1749,12 @@ TM_Result UHeapDelete(Relation relation, ItemPointer tid, CommandId cid, Snapsho
RowPtr *rp = UPageGetRowPtr(page, offnum);
Assert(RowPtrIsNormal(rp) || RowPtrIsDeleted(rp));
/* prune FSM and also the page, may delete some utuples */
UHeapPagePruneFSM(relation, buffer, fxid, page, blkno);
UHeapResetWaitTimeForTDSlot();
/* check the status of the utuple */
check_tup_satisfies_update:
result = UHeapTupleSatisfiesUpdate(relation, snapshot, tid, &utuple, cid, buffer, &ctid, &tdinfo, &updateSubXid,
&lockerXid, &lockerSubXid, false, multixidIsMyself, &inplaceUpdatedOrLocked, allowDeleteSelf);
@ -1798,6 +1838,7 @@ check_tup_satisfies_update:
SubXactLockTableInsert(subxid);
}
/* reserve a transaction slot on the page */
transSlotId = UHeapPageReserveTransactionSlot(relation, buffer, fxid,
&prevUrecptr, &lockReacquired, InvalidBuffer, &minXidInTDSlots);
@ -1882,6 +1923,7 @@ check_tup_satisfies_update:
Oid relOid = RelationIsPartition(relation) ? GetBaseRelOidOfParition(relation) : RelationGetRelid(relation);
Oid partitionOid = RelationIsPartition(relation) ? RelationGetRelid(relation) : InvalidOid;
/* prepare infomation for undo delete, this include the information of the deleted utuple */
urecptr = UHeapPrepareUndoDelete(relOid, partitionOid, RelationGetRelFileNode(relation),
RelationGetRnodeSpace(relation), persistence, buffer, offnum, fxid, subxid, cid,
prevUrecptr, INVALID_UNDO_REC_PTR, &oldTD, &utuple, InvalidBlockNumber, NULL, &xlum);
@ -2031,12 +2073,26 @@ void PutBlockInplaceUpdateTuple(Page page, Item item, RowPtr *lp, Size size)
/*
* UHeapUpdate - update a tuple
*
* This function either updates the tuple in-place or it deletes the old
* tuple and new tuple for non-in-place updates. Additionally this function
* inserts an undo record and updates the undo pointer in page header.
*
* For input and output values, see heap_update.
* @param[IN] relation: relation the tuple is in.
* @param[IN] parentRelation: parent relation of the tuple.
* @param[IN] otid: TID of the tuple to update.
* @param[IN] newtup: the new tuple data.
* @param[IN] cid: CID of the updating transaction.
* @param[IN] crosscheck: snapshot to use for checking old tuple's visibility.
* @param[IN] snapshot: snapshot of the current transaction.
* @param[IN] wait: if true, wait for conflicting transactions to end.
* @param[OUT] oldslot: slot to store the old tuple in.
* @param[OUT] tmfd: filled with info about the old tuple.
* @param[OUT] indexkey_update_flag: filled with true if any of the updated columns are index keys.
* @param[OUT] modifiedIdxAttrs: filled with the bitmap of index attributes that are modified.
* @param[IN] allow_inplace_update: if true, allow update in place.
* @return: TM_Result
* @see: heap_update
* @note: This function either updates the tuple in-place or it deletes the old
* tuple and new tuple for non-in-place updates. Additionally this function
* inserts an undo record and updates the undo pointer in page header.
*
* For input and output values, see heap_update.
*/
TM_Result UHeapUpdate(Relation relation, Relation parentRelation, ItemPointer otid, UHeapTuple newtup, CommandId cid,
Snapshot crosscheck, Snapshot snapshot, bool wait, TupleTableSlot **oldslot, TM_FailureData *tmfd,
@ -2132,6 +2188,7 @@ TM_Result UHeapUpdate(Relation relation, Relation parentRelation, ItemPointer ot
block = ItemPointerGetBlockNumber(otid);
page = GetPageBuffer(relation, block, buffer);
// interestingAttrs contains all columns of the index.
interestingAttrs = NULL;
interestingAttrs = bms_add_members(interestingAttrs, inplaceUpdAttrs);
interestingAttrs = bms_add_members(interestingAttrs, keyAttrs);
@ -2140,6 +2197,7 @@ TM_Result UHeapUpdate(Relation relation, Relation parentRelation, ItemPointer ot
LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
// Fetch the tuple to be updated.
oldOffnum = ItemPointerGetOffsetNumber(otid);
lp = UPageGetRowPtr(page, oldOffnum);
Assert(RowPtrIsNormal(lp) || RowPtrIsDeleted(lp));
@ -2151,6 +2209,11 @@ check_tup_satisfies_update:
lockerRemains = false;
anyMultiLockerMemberAlive = true;
/* Check whether the tuple is visible to us.
* After the function returned, the oldtup is filled with the tuple data.
* The inplaceUpdated is set to true if the tuple is inplace updated.
* The lockerXid is set to the locker xid if the tuple is locked by other transaction.
*/
result = UHeapTupleSatisfiesUpdate(relation, snapshot, otid, &oldtup, cid, buffer, &ctid, &txactinfo, &updateSubXid,
&lockerXid, &lockerSubXid, false, multixidIsMyself, &inplaceUpdated);
@ -2197,6 +2260,7 @@ check_tup_satisfies_update:
if (!UHeapWait(relation, buffer, &oldtup, lockmode, false, txactinfo.xid, lockerXid, updateSubXid,
lockerSubXid, &haveTupleLock, &multixidIsMyself)) {
//we need to recheck the wirte-write conflict
goto check_tup_satisfies_update;
}
@ -2222,6 +2286,7 @@ check_tup_satisfies_update:
* for the same page.
*/
lp = UPageGetRowPtr(page, oldOffnum);
// if result is not TM_Ok, then we need to fill the tmfd and return the result
if (result != TM_Ok) {
Assert(result == TM_SelfModified || result == TM_SelfUpdated || result == TM_Updated || result == TM_Deleted ||
result == TM_BeingModified);
@ -2254,6 +2319,7 @@ check_tup_satisfies_update:
bms_free(inplaceUpdAttrs);
bms_free(keyAttrs);
// if newtup is not inplace updated or there exists modified attributes, then the indexkey_update_flag is true
*indexkey_update_flag = !UHeapTupleIsInPlaceUpdated(((UHeapTuple)newtup)->disk_tuple->flag) ||
(modifiedIdxAttrs != NULL && *modifiedIdxAttrs != NULL);
@ -2268,6 +2334,9 @@ check_tup_satisfies_update:
newtup->table_oid = RelationGetRelid(relation);
newtup->xc_node_id = u_sess->pgxc_cxt.PGXCNodeIdentifier;
/* If modifiedAttrs and inplaceUpdAttrs have intersection, then the index attributes are updated.
* In this case, we need to set the modifiedIdxAttrs to the intersection of modifiedAttrs and inplaceUpdAttrs.
*/
isIndexUpdated = bms_overlap(modifiedAttrs, inplaceUpdAttrs);
if (modifiedIdxAttrs != NULL) {
*modifiedIdxAttrs = isIndexUpdated ? bms_intersect(modifiedAttrs, inplaceUpdAttrs) : NULL;
@ -2323,6 +2392,7 @@ check_tup_satisfies_update:
}
}
} else if (newtupsize <= oldtupsize) {
// If newtuple is smaller than oldtuple, then we can do inplace update.
useInplaceUpdate = true;
}
@ -2463,6 +2533,9 @@ check_tup_satisfies_update:
* contention on transaction slots.
*/
if (!needToast) {
/* Since we can not do inplace update, so we need to find a new page to store the new tuple.
* We try to acquire new page by calling RelationGetBufferForUTuple.
*/
newbuf = RelationGetBufferForUTuple(relation, uheaptup->disk_tuple_size, buffer, 0, NULL);
} else {
/* Re-acquire the lock on the old tuple's page. */
@ -2614,7 +2687,9 @@ check_tup_satisfies_update:
uheaptup = newtup;
}
/* Till now, we know whether we will delete the old index */
/* Till now, we know whether we will delete the old index.
* If the index attributes are updated or the tuple is non-inplace updating, then we will delete the old index tuple.
*/
if (oldslot && (*modifiedIdxAttrs != NULL || !useInplaceUpdate)) {
*oldslot = MakeSingleTupleTableSlot(relation->rd_att, false, TAM_USTORE);
TupleDesc rowDesc = (*oldslot)->tts_tupleDescriptor;
@ -2653,6 +2728,7 @@ check_tup_satisfies_update:
uint16 prefixlen = 0;
uint16 suffixlen = 0;
uint8 xorDeltaFlags = 0;
// oldp and newp point to the data of the tuple, oldlen and newlen are the length of the tuple data.
char *oldp = (char *)oldtup.disk_tuple + oldtup.disk_tuple->t_hoff;
char *newp = (char *)uheaptup->disk_tuple + uheaptup->disk_tuple->t_hoff;
int oldlen = oldtup.disk_tuple_size - oldtup.disk_tuple->t_hoff;
@ -2665,6 +2741,7 @@ check_tup_satisfies_update:
oldpTmp = oldp;
newpTmp = newp;
// calculate the prefixlen and suffixlen
for (prefixlen = 0; prefixlen < minlen; prefixlen++, oldpTmp++, newpTmp++) {
if (*oldpTmp != *newpTmp) {
break;
@ -2696,7 +2773,7 @@ check_tup_satisfies_update:
if (suffixlen > 0)
undoXorDeltaSize += sizeof(uint16);
}
/* The undoXorDeltaSize is the length of the delta data stored in the undo record. */
/* The first sizeof(uint8) is space for t_hoff and the second sizeof(uint8) is space for prefix and suffix flag
*/
undoXorDeltaSize += sizeof(uint8) + oldtup.disk_tuple->t_hoff - OffsetTdId + sizeof(uint8);
@ -2714,6 +2791,7 @@ check_tup_satisfies_update:
UndoRecord *undorec = (*u_sess->ustore_cxt.urecvec)[0];
if (useInplaceUpdate) {
// fill the undo record data
appendBinaryStringInfo(undorec->Rawdata(), (char *)&(oldtup.disk_tuple->t_hoff), sizeof(uint8));
appendBinaryStringInfo(undorec->Rawdata(), (char *)oldtup.disk_tuple + OffsetTdId,
oldtup.disk_tuple->t_hoff - OffsetTdId);

7
src/gausskernel/storage/access/ustore/knl_upage.cpp
View File

@ -187,6 +187,11 @@ static void FindNextFreeSlot(const UHeapBufferPage *bufpage, Page input_page, Of
}
}
/*
* Do the actual insert of the item into the page
* Put the item data on the back or the page and update the page's lower
* and upper pointers.
*/
static bool CalculateLowerUpperPointers(Page page, OffsetNumber offsetNumber, Item item, Size size, bool needshuffle)
{
int lower;
@ -290,6 +295,7 @@ OffsetNumber UPageAddItem(Relation rel, UHeapBufferPage *bufpage, Item item, Siz
return InvalidOffsetNumber;
}
} else {
// Find free slot in the page
FindNextFreeSlot(bufpage, input_page, &offsetNumber);
}
@ -307,6 +313,7 @@ OffsetNumber UPageAddItem(Relation rel, UHeapBufferPage *bufpage, Item item, Siz
return InvalidOffsetNumber;
}
// do the actual insert of the item
if (!CalculateLowerUpperPointers(page, offsetNumber, item, size, needshuffle)) {
return InvalidOffsetNumber;
}