diff --git a/src/gausskernel/storage/access/redo/redo_ubtxlog.cpp b/src/gausskernel/storage/access/redo/redo_ubtxlog.cpp index 74af9091e8c92342e099f09c72ff329fcedd5a39..4ecd63224d2d631200d0972fa27f9d7497ebd4b2 100644 --- a/src/gausskernel/storage/access/redo/redo_ubtxlog.cpp +++ b/src/gausskernel/storage/access/redo/redo_ubtxlog.cpp @@ -1621,6 +1621,21 @@ void UBTree3XlogDeleteOperatorPage(RedoBufferInfo* buffer, void* recorddata, Und PageSetLSN(page, buffer->lsn); } +void UBTree3XlogFreezeTdOperatorPage(RedoBufferInfo* buffer, void* recorddata) +{ + xl_ubtree3_freeze_td_slot *xlrec = (xl_ubtree3_freeze_td_slot *)recorddata; + uint8 *frozenSlots = (uint8 *)(((char*)recorddata) + SizeOfUbtree3FreezeTDSlot); + Buffer buf = buffer->buf; + Page page = BufferGetPage(buf); + UBTreeFreezeOrInvalidIndexTuples(buf, xlrec->nFrozen, frozenSlots, true); + UBTPCRPageOpaque pageop = (UBTPCRPageOpaque)PageGetSpecialPointer(page); + for (uint16 i = 0; i < xlrec->nFrozen; i++) { + UBTreeTD curTd = (UBTreeTD)UBTreePCRGetTD(page, frozenSlots[i] + 1); + curTd->setFrozen(); + } + PageSetLSN(page, buffer->lsn); +} + void UBTree3XlogReuseTdOperatorPage(RedoBufferInfo* buffer, void* recorddata) { uint8 *ncompletedXactSlots = (uint8 *)recorddata; @@ -1639,6 +1654,33 @@ void UBTree3XlogReuseTdOperatorPage(RedoBufferInfo* buffer, void* recorddata) UBTreePCRTDSetStatus(curTd, TD_COMMITED); Assert(TransactionIdIsValid(curTd->xactid)); } + PageSetLSN(page, buffer->lsn); +} + +void UBTree3XlogExtendTdOperatorPage(RedoBufferInfo* buffer, void* recorddata) +{ + xl_ubtree3_extend_td_slots *xlrec = (xl_ubtree3_extend_td_slots *)recorddata; + Page page = buffer->pageinfo.page; + PageHeader phdr = (PageHeader)page; + UBTPCRPageOpaque op = (UBTPCRPageOpaque)PageGetSpecialPointer(page); + uint8 curTdSlots = op->td_count; + Assert(curTdSlots == xlrec->nPrevSlots); + + char *start = ((char *)page) + UBTreePCRGetRowPtrOffset(page); + char *end = page + phdr->pd_lower; + size_t lpSize = end - start; + errno_t rc = memmove_s((char *)start + (xlrec->nExtended * sizeof(UBTreeTDData)), lpSize, start, lpSize); + securec_check(rc, "", ""); + + for (int i = curTdSlots; i < curTdSlots + xlrec->nExtended; i++) { + UBTreeTD curTd = (UBTreeTD)UBTreePCRGetTD(page, i + 1); + rc = memset_s(curTd, sizeof(UBTreeTDData), 0, sizeof(UBTreeTDData)); + securec_check(rc, "", ""); + } + + op->td_count += xlrec->nExtended; + phdr->pd_lower += xlrec->nExtended * sizeof(UBTreeTDData); + PageSetLSN(page, buffer->lsn); } void UBTree3XlogRollbackTxnOperatorPage(RedoBufferInfo* buffer, void* recorddata) diff --git a/src/gausskernel/storage/access/ubtree/ubtxlog.cpp b/src/gausskernel/storage/access/ubtree/ubtxlog.cpp index 9d9aea11c58f43405035b5d9327ac03359b72a6c..604e282feb1053c9b66ed8de76d9d79bd1f91e98 100644 --- a/src/gausskernel/storage/access/ubtree/ubtxlog.cpp +++ b/src/gausskernel/storage/access/ubtree/ubtxlog.cpp @@ -980,8 +980,6 @@ static UndoRecPtr PrepareUndoRecordForRedo(XLogReaderState *record, /* recover undo record */ UndoRecord *undorec = (*t_thrd.ustore_cxt.urecvec)[0]; undorec->SetUrp(urecptr); - undorec->SetBlkprev(prevurp); - undorec->SetOldXactId(xlrec->prevXidOfTuple); /* We need to pass in tablespace and relfilenode in PrepareUndo but we never explicitly * wrote those information in the xlundohdr because we can grab them from the XLOG record itself. @@ -993,18 +991,19 @@ static UndoRecPtr PrepareUndoRecordForRedo(XLogReaderState *record, if (isInsert) { urecptr = UBTreePCRPrepareUndoInsert(xlundohdr->relOid, partitionOid, targetNode.relNode, - targetNode.spcNode, UNDO_PERMANENT, xlrec->curXid, FirstCommandId, - blkprev, prevurp, blkno, xlundohdr, &undometa, InvalidOffsetNumber, InvalidBuffer, - false, uinfo, itup); - + targetNode.spcNode, UNDO_PERMANENT, xlrec->curXid, FirstCommandId, + blkprev, prevurp, blkno, xlundohdr, &undometa, InvalidOffsetNumber, InvalidBuffer, + xlrec->prevXidOfTuple, uinfo, itup); } else { urecptr = UBTreePCRPrepareUndoDelete(xlundohdr->relOid, partitionOid, targetNode.relNode, - targetNode.spcNode, UNDO_PERMANENT, InvalidBuffer, xlrec->curXid, 0, prevurp, - itup, blkno, xlundohdr, &undometa, uinfo); + targetNode.spcNode, UNDO_PERMANENT, InvalidBuffer, xlrec->curXid, 0, prevurp, + itup, blkno, xlundohdr, &undometa, uinfo); } Assert(UNDO_PTR_GET_OFFSET(urecptr) == UNDO_PTR_GET_OFFSET(xlundohdr->urecptr)); undorec->SetOffset(xlrec->offNum); + undorec->SetBlkprev(blkprev); + undorec->SetOldXactId(xlrec->prevXidOfTuple); if (!skipInsert) { /* Insert the Undo record into the undo store */ InsertPreparedUndo(t_thrd.ustore_cxt.urecvec, lsn); @@ -1260,8 +1259,10 @@ static void UBTree3RestoreMeta(XLogReaderState *record, uint8 block_id) ptr = XLogRecGetBlockData(record, block_id, &len); UBTree3RestoreMetaOperatorPage(&metabuf, (void *)ptr, len); - MarkBufferDirty(metabuf.buf); - UnlockReleaseBuffer(metabuf.buf); + if (BufferIsValid(metabuf.buf)) { + MarkBufferDirty(metabuf.buf); + UnlockReleaseBuffer(metabuf.buf); + } } static void UBTree3XlogInsertPcrInternal(XLogReaderState* record, bool hasMeta) @@ -1269,6 +1270,16 @@ static void UBTree3XlogInsertPcrInternal(XLogReaderState* record, bool hasMeta) RedoBufferInfo lbuf; RedoBufferInfo cbuf; + if (XLogReadBufferForRedo(record, UBTREE3_INSERT_PCR_LEAF_BLOCK_NUM, &cbuf) == BLK_NEEDS_REDO) { + UBTreeXlogClearIncompleteSplit(&cbuf); + if (BufferIsValid(cbuf.buf)) { + MarkBufferDirty(cbuf.buf); + } + } + if (BufferIsValid(cbuf.buf)) { + UnlockReleaseBuffer(cbuf.buf); + } + if (XLogReadBufferForRedo(record, UBTREE3_INSERT_PCR_INTERNAL_BLOCK_NUM, &lbuf) == BLK_NEEDS_REDO) { Size dataLen; char *dataPos = XLogRecGetBlockData(record, BTREE_SPLIT_LEFT_BLOCK_NUM, &dataLen); @@ -1280,25 +1291,17 @@ static void UBTree3XlogInsertPcrInternal(XLogReaderState* record, bool hasMeta) errmsg("faild to add left P_HIKEY wihle redo."))); } PageSetLSN(page, lbuf.lsn); + if (BufferIsValid(lbuf.buf)) { + MarkBufferDirty(lbuf.buf); + } } if (BufferIsValid(lbuf.buf)) { - MarkBufferDirty(lbuf.buf); UnlockReleaseBuffer(lbuf.buf); } - if (XLogReadBufferForRedo(record, UBTREE3_INSERT_PCR_LEAF_BLOCK_NUM, &cbuf) == BLK_NEEDS_REDO) { - UBTreeXlogClearIncompleteSplit(&cbuf); - } - if (BufferIsValid(cbuf.buf)) { - MarkBufferDirty(cbuf.buf); - UnlockReleaseBuffer(cbuf.buf); - } - if (hasMeta) { UBTree3RestoreMeta(record, UBTREE3_INSERT_PCR_META_BLOCK_NUM); } - - return; } static void UBTree3XlogInsert(XLogReaderState* record, bool isDup) @@ -1311,8 +1314,8 @@ static void UBTree3XlogInsert(XLogReaderState* record, bool isDup) recordptr += IndexTupleSize(itup); bool replayAll = !AmPageRedoWorker() || !SUPPORT_USTORE_UNDO_WORKER; bool replayRedoOnly = replayAll ? false : parallel_recovery::DoPageRedoWorkerReplayUndo(); - UndoRecPtr urec = PrepareUndoRecordForRedo(record, xlrec, itup, recordptr, (replayAll | replayRedoOnly), true); - if (replayAll | !replayRedoOnly) { + UndoRecPtr urec = PrepareUndoRecordForRedo(record, xlrec, itup, recordptr, (replayAll || replayRedoOnly), true); + if (replayAll || !replayRedoOnly) { if (XLogReadBufferForRedo(record, BTREE_INSERT_ORIG_BLOCK_NUM, &buffer) == BLK_NEEDS_REDO) { Page page = buffer.pageinfo.page; if (!isDup) { @@ -1333,14 +1336,14 @@ static void UBTree3XlogInsert(XLogReaderState* record, bool isDup) UBTreePCRClearIndexTupleDeleted(iid); } PageSetLSN(page, buffer.lsn); + if (BufferIsValid(buffer.buf)) { + MarkBufferDirty(buffer.buf); + } } if (BufferIsValid(buffer.buf)) { - MarkBufferDirty(buffer.buf); UnlockReleaseBuffer(buffer.buf); } } - - return; } static void UBTree3XlogNewRoot(XLogReaderState* record) @@ -1369,14 +1372,14 @@ static void UBTree3XlogNewRoot(XLogReaderState* record) if (UBTPCRPageAddItem(page, lItem, lItemSz, P_HIKEY, false) == InvalidOffsetNumber) { ereport(PANIC, (errcode(ERRCODE_INDEX_CORRUPTED), - errmsg("faild to add left P_HIKEY wihle redo."))); + errmsg("faild to add left P_HIKEY wihle redo."))); } if (UBTPCRPageAddItem(page, rItem, rItemSz, P_FIRSTKEY, false) == InvalidOffsetNumber) { ereport(PANIC, (errcode(ERRCODE_INDEX_CORRUPTED), - errmsg("faild to add left P_FIRSTKEY wihle redo."))); + errmsg("faild to add left P_FIRSTKEY wihle redo."))); } - PageSetLSN(page, buffer.lsn); } + PageSetLSN(page, buffer.lsn); if (BufferIsValid(buffer.buf)) { MarkBufferDirty(buffer.buf); @@ -1387,13 +1390,14 @@ static void UBTree3XlogNewRoot(XLogReaderState* record) lbuffer.buf = InvalidBuffer; if (xlrec->level > 0 && XLogReadBufferForRedo(record, 1, &lbuffer) == BLK_NEEDS_REDO) { UBTreeXlogClearIncompleteSplit(&lbuffer); + if (BufferIsValid(lbuffer.buf)) { + MarkBufferDirty(lbuffer.buf); + } } if (BufferIsValid(lbuffer.buf)) { - MarkBufferDirty(lbuffer.buf); UnlockReleaseBuffer(lbuffer.buf); } - - return; + UBTree3RestoreMeta(record, BTREE_NEWROOT_META_BLOCK_NUM); } static void UBTree3XlogPrunePage(XLogReaderState* record) @@ -1472,7 +1476,7 @@ static void UBTree3XlogFreezeTdSlot(XLogReaderState* record) } if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO) { - UBTree3XlogPrunePageOperatorPage(&buffer, XLogRecGetData(record), rnode.relNode); + UBTree3XlogFreezeTdOperatorPage(&buffer, (void *)XLogRecGetData(record)); if (BufferIsValid(buffer.buf)) { MarkBufferDirty(buffer.buf); } @@ -1481,7 +1485,6 @@ static void UBTree3XlogFreezeTdSlot(XLogReaderState* record) if (BufferIsValid(buffer.buf)) { UnlockReleaseBuffer(buffer.buf); } - return; } static void UBTree3XlogReuseTdSlot(XLogReaderState* record) @@ -1497,14 +1500,13 @@ static void UBTree3XlogReuseTdSlot(XLogReaderState* record) if (BufferIsValid(buffer.buf)) { UnlockReleaseBuffer(buffer.buf); } - return; } static void UBTree3XlogExtendTdSlots(XLogReaderState* record) { RedoBufferInfo buffer; if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO) { - UBTree3XlogReuseTdOperatorPage(&buffer, (void *)XLogRecGetData(record)); + UBTree3XlogExtendTdOperatorPage(&buffer, (void *)XLogRecGetData(record)); if (BufferIsValid(buffer.buf)) { MarkBufferDirty(buffer.buf); } @@ -1513,7 +1515,6 @@ static void UBTree3XlogExtendTdSlots(XLogReaderState* record) if (BufferIsValid(buffer.buf)) { UnlockReleaseBuffer(buffer.buf); } - return; } static void UBTree3XlogRollBackTxn(XLogReaderState* record) @@ -1531,24 +1532,25 @@ static void UBTree3XlogRollBackTxn(XLogReaderState* record) } } -void BTree3XlogSplitLeftPage(RedoBufferInfo *bufferinfo, void *recorddata, bool onleft, void *blkdata, Size datalen) +void UBTree3XlogSplitLeftPage(RedoBufferInfo *bufferinfo, void *recorddata, bool onleft, void *blkdata, Size datalen) { xl_ubtree3_split *xlrec = (xl_ubtree3_split *)recorddata; Page lpage = bufferinfo->pageinfo.page; UBTPCRPageOpaque lopaque = (UBTPCRPageOpaque)PageGetSpecialPointer(lpage); char *dataPos = (char *)blkdata; IndexTuple item; - Size size; + Size size = 0; if (onleft) { item = (IndexTuple)dataPos; size = MAXALIGN(IndexTupleSize(item)); dataPos += size; datalen -= size; } - Item leftHightKey = (Item)dataPos; - Size leftHightKeySz = MAXALIGN(IndexTupleSize(leftHightKey)); - dataPos += leftHightKeySz; - dataPos -= leftHightKeySz; + Item leftHighKey = (Item)dataPos; + Size leftHighKeySz = MAXALIGN(IndexTupleSize(leftHighKey)); + dataPos += leftHighKeySz; + datalen -= leftHighKeySz; + Assert(datalen == 0); START_CRIT_SECTION(); Page lNewPage = PageGetTempPageCopySpecial(lpage); @@ -1557,13 +1559,13 @@ void BTree3XlogSplitLeftPage(RedoBufferInfo *bufferinfo, void *recorddata, bool } END_CRIT_SECTION(); OffsetNumber leftOff = P_HIKEY; - if (UBTPCRPageAddItem(lNewPage, leftHightKey, leftHightKeySz, P_HIKEY, false) == InvalidOffsetNumber) { + if (UBTPCRPageAddItem(lNewPage, leftHighKey, leftHighKeySz, P_HIKEY, false) == InvalidOffsetNumber) { ereport(PANIC, (errcode(ERRCODE_INDEX_CORRUPTED), errmsg("faild to add left P_HIKEY wihle redo."))); } leftOff = OffsetNumberNext(leftOff); OffsetNumber off; - for (off = P_FIRSTDATAKEY(lopaque); off < xlrec->firstRight; off ++) { + for (off = P_FIRSTDATAKEY(lopaque); off < xlrec->firstRight; off++) { if (onleft && off == xlrec->newItemOff) { if (!UBTreePCRPageAddTuple(lNewPage, size, NULL, item, leftOff, false, xlrec->slotNo)) { ereport(ERROR, (errcode(ERRCODE_INDEX_CORRUPTED), errmsg("failed to add new item to block"))); @@ -1574,7 +1576,7 @@ void BTree3XlogSplitLeftPage(RedoBufferInfo *bufferinfo, void *recorddata, bool UBTreeItemId iid = UBTreePCRGetRowPtr(lpage, off); IndexTuple itup = UBTreePCRGetIndexTuple(lpage, off); Size itupSize = IndexTupleSize(itup); - if (!UBTreePCRPageAddTuple(lNewPage, itupSize, iid, itup, leftOff, true, UBTreeInvalidTDSlotId)) { + if (!UBTreePCRPageAddTuple(lNewPage, itupSize, iid, itup, leftOff, true, UBTreeInvalidTDSlotId)) { ereport(ERROR, (errcode(ERRCODE_INDEX_CORRUPTED), errmsg("failed to add new item to block"))); } leftOff = OffsetNumberNext(leftOff); @@ -1617,7 +1619,7 @@ static void UBTree3XlogSplit(XLogReaderState* record, bool onLeft, bool isRoot) rec += SizeOfUbtree3InsertOrDelete; IndexTuple itup = (IndexTuple)rec; rec += IndexTupleSize(itup); - xlrec->urp = PrepareUndoRecordForRedo(record, xlrecInsert, itup, rec, (replayAll | replayRedoOnly), true); + xlrec->urp = PrepareUndoRecordForRedo(record, xlrecInsert, itup, rec, (replayAll || replayRedoOnly), true); } if (replayRedoOnly) { @@ -1628,9 +1630,11 @@ static void UBTree3XlogSplit(XLogReaderState* record, bool onLeft, bool isRoot) RedoBufferInfo cbuf; if (XLogReadBufferForRedo(record, BTREE_SPLIT_CHILD_BLOCK_NUM, &cbuf) == BLK_NEEDS_REDO) { UBTreeXlogClearIncompleteSplit(&cbuf); + if (BufferIsInvalid(cbuf.buf)) { + MarkBufferDirty(cbuf.buf); + } } if (BufferIsInvalid(cbuf.buf)) { - MarkBufferDirty(cbuf.buf); UnlockReleaseBuffer(cbuf.buf); } } @@ -1650,34 +1654,38 @@ static void UBTree3XlogSplit(XLogReaderState* record, bool onLeft, bool isRoot) PageSetLSN(rpage, rbuf.lsn); if (BufferIsValid(rbuf.buf)) { MarkBufferDirty(rbuf.buf); - UnlockReleaseBuffer(rbuf.buf); } RedoBufferInfo lbuf; if (XLogReadBufferForRedo(record, BTREE_SPLIT_LEFT_BLOCK_NUM, &lbuf) == BLK_NEEDS_REDO) { dataPos = XLogRecGetBlockData(record, BTREE_SPLIT_LEFT_BLOCK_NUM, &dataLen); - BTree3XlogSplitLeftPage(&lbuf, (void *)xlrec, onLeft, (void *)dataPos, dataLen); + UBTree3XlogSplitLeftPage(&lbuf, (void *)xlrec, onLeft, (void *)dataPos, dataLen); + if (BufferIsValid(lbuf.buf)) { + MarkBufferDirty(lbuf.buf); + } } if (BufferIsValid(lbuf.buf)) { - MarkBufferDirty(lbuf.buf); UnlockReleaseBuffer(lbuf.buf); } + if (BufferIsValid(rbuf.buf)) { + UnlockReleaseBuffer(rbuf.buf); + } RedoBufferInfo sbuf; if (rnext != P_NONE) { - if (XLogReadBufferForRedo(record, BTREE_SPLIT_LEFT_BLOCK_NUM, &sbuf) == BLK_NEEDS_REDO) { + if (XLogReadBufferForRedo(record, BTREE_SPLIT_RIGHTNEXT_BLOCK_NUM, &sbuf) == BLK_NEEDS_REDO) { Page page = sbuf.pageinfo.page; UBTPCRPageOpaque opaque = (UBTPCRPageOpaque)PageGetSpecialPointer(page); opaque->btpo_prev = rightSib; PageSetLSN(page, sbuf.lsn); + if (BufferIsValid(sbuf.buf)) { + MarkBufferDirty(sbuf.buf); + } + } + if (BufferIsValid(sbuf.buf)) { + UnlockReleaseBuffer(sbuf.buf); } } - if (BufferIsValid(sbuf.buf)) { - MarkBufferDirty(sbuf.buf); - UnlockReleaseBuffer(sbuf.buf); - } - - return; } static void UBTree4XlogUnlinkPage(uint8 info, XLogReaderState *record) @@ -1724,8 +1732,10 @@ static void UBTree4XlogUnlinkPage(uint8 info, XLogReaderState *record) XLogInitBufferForRedo(record, BTREE_UNLINK_PAGE_CUR_PAGE_NUM, &buffer); UBTree4XlogUnlinkPageOperatorCurpage(&buffer, xlrec); - MarkBufferDirty(buffer.buf); - UnlockReleaseBuffer(buffer.buf); + if (BufferIsValid(buffer.buf)) { + MarkBufferDirty(buffer.buf); + UnlockReleaseBuffer(buffer.buf); + } /* * If we deleted a parent of the targeted leaf page, instead of the leaf @@ -1740,9 +1750,10 @@ static void UBTree4XlogUnlinkPage(uint8 info, XLogReaderState *record) RedoBufferInfo cbuffer; XLogInitBufferForRedo(record, BTREE_UNLINK_PAGE_CHILD_NUM, &cbuffer); UBTree4XlogUnlinkPageOperatorChildpage(&cbuffer, xlrec); - - MarkBufferDirty(cbuffer.buf); - UnlockReleaseBuffer(cbuffer.buf); + if (BufferIsValid(cbuffer.buf)) { + MarkBufferDirty(cbuffer.buf); + UnlockReleaseBuffer(cbuffer.buf); + } } /* Update metapage if needed */ @@ -1791,8 +1802,9 @@ static void UBTree4XlogMarkPageHalfDead(XLogReaderState *record) MarkBufferDirty(lbuffer.buf); } } - - UnlockReleaseBuffer(lbuffer.buf); + if (BufferIsValid(lbuffer.buf)) { + UnlockReleaseBuffer(lbuffer.buf); + } } void UBTree3Redo(XLogReaderState* record) diff --git a/src/gausskernel/storage/access/ubtreepcr/ubtpcrinsert.cpp b/src/gausskernel/storage/access/ubtreepcr/ubtpcrinsert.cpp index e4bef828d1d74e1155b795a04d4ceaec0ab00bcc..9e8f50182aae83e9100c51506b7930ddb0053dd3 100644 --- a/src/gausskernel/storage/access/ubtreepcr/ubtpcrinsert.cpp +++ b/src/gausskernel/storage/access/ubtreepcr/ubtpcrinsert.cpp @@ -331,13 +331,12 @@ top: UndoPersistence persistence = UndoPersistenceForRelation(rel); Oid relOid = RelationIsPartition(rel) ? GetBaseRelOidOfParition(rel) : RelationGetRelid(rel); Oid partitionOid = RelationIsPartition(rel) ? RelationGetRelid(rel) : InvalidOid; - bool selfInsert = prevTd.xactid == fxid; undoinfo.prev_td_id = prevSlot; UndoRecPtr urecPtr = INVALID_UNDO_REC_PTR; urecPtr = UBTreePCRPrepareUndoInsert(relOid, partitionOid, RelationGetRelFileNode(rel), - RelationGetRnodeSpace(rel), persistence, GetTopTransactionId(), GetCurrentCommandId(true), + RelationGetRnodeSpace(rel), persistence, fxid, GetCurrentCommandId(true), prevTd.undoRecPtr, INVALID_UNDO_REC_PTR, InvalidBlockNumber, NULL, &xlum, offset, InvalidBuffer, - selfInsert, &undoinfo, itup); + (prevTd.xactid == fxid ? fxid : FrozenTransactionId), &undoinfo, itup); if (isNormalInsert) { split = UBTreePCRInsertOnPage(rel, itupKey, buf, InvalidBuffer, stack, itup, @@ -2244,8 +2243,8 @@ static Buffer UBTreePCRSplit(Relation rel, Buffer buf, Buffer cbuf, OffsetNumber walInfo->itup = (IndexTuple)((char*)(cur_urec->Rawdata()->data + SizeOfUBTreeUndoInfoData)); walInfo->flag = flag; } - LogSplit(buf, rbuf, sbuf, cbuf, firstrightoff, newitemoff, newitemleftoff, - leftoff, tdslot, walInfo, newitemonleft); + LogSplit(buf, rbuf, sbuf, cbuf, firstrightoff, isroot, newitemoff, newitemleftoff, + tdslot, walInfo, newitemonleft); if (walInfo != NULL) { pfree(walInfo); @@ -2467,7 +2466,7 @@ uint8 PreparePCRDelete(Relation rel, Buffer buf, OffsetNumber offnum, UBTreeUndo return UBTreeInvalidTDSlotId; } - /* Get tuple's previouse operation's xact info */ + /* Get tuple's previous operation's xact info */ Page page = BufferGetPage(buf); UBTreeItemId iid = (UBTreeItemId)UBTreePCRGetRowPtr(page, offnum); uint8 slotNo = iid->lp_td_id; @@ -2857,6 +2856,16 @@ void LogSplit(Buffer buf, Buffer rbuf, Buffer sbuf, Buffer leftCbuf, OffsetNumbe xl_ubtree3_split xlrecSplit; uint8 xlflag; + if (walInfo != NULL) { + xlrecInsert.offNum = originOff; + xlrecInsert.tdId = walInfo->tdId; + xlrecInsert.prevXidOfTuple = walInfo->oldXid; + xlrecInsert.curXid = walInfo->xid; + xlundohdr.relOid = walInfo->relOid; + xlundohdr.urecptr = walInfo->urecptr; + xlundohdr.flag = walInfo->flag; + } + Page leftpage = BufferGetPage(buf); Page rightpage = BufferGetPage(rbuf); UBTPCRPageOpaque lopaque = (UBTPCRPageOpaque)PageGetSpecialPointer(leftpage); @@ -2870,7 +2879,8 @@ void LogSplit(Buffer buf, Buffer rbuf, Buffer sbuf, Buffer leftCbuf, OffsetNumbe xlrecSplit.slotNo = tdSlot; if (newItemOnLeft) { - UBTreeTD td = UBTreePCRGetTD(leftpage, tdSlot); + UBTreeItemId iid = UBTreePCRGetRowPtr(leftpage, leftOff); + UBTreeTD td = UBTreePCRGetTD(leftpage, iid->lp_td_id); xlrecSplit.fxid = td->xactid; xlrecSplit.urp = td->undoRecPtr; } @@ -2879,14 +2889,6 @@ void LogSplit(Buffer buf, Buffer rbuf, Buffer sbuf, Buffer leftCbuf, OffsetNumbe XLogRegisterData((char *)&xlrecSplit, SizeOfUbtree3Split); if (walInfo != NULL) { - xlrecInsert.offNum = originOff; - xlrecInsert.tdId = walInfo->tdId; - xlrecInsert.prevXidOfTuple = walInfo->oldXid; - xlrecInsert.curXid = walInfo->xid; - xlundohdr.relOid = walInfo->relOid; - xlundohdr.urecptr = walInfo->urecptr; - xlundohdr.flag = walInfo->flag; - XLogRegisterData((char *)&xlrecInsert, SizeOfUbtree3InsertOrDelete); XLogRegisterData((char *)walInfo->itup, IndexTupleSize(walInfo->itup)); XLogRegisterData((char *)&xlundohdr, SizeOfXLUndoHeader); @@ -2916,7 +2918,7 @@ void LogSplit(Buffer buf, Buffer rbuf, Buffer sbuf, Buffer leftCbuf, OffsetNumbe if (newItemOnLeft) { IndexTuple itup = UBTreePCRGetIndexTuple(leftpage, leftOff); - XLogRegisterBufData(0, (char *)itup, IndexTupleSize(itup)); + XLogRegisterBufData(0, (char *)itup, MAXALIGN(IndexTupleSize(itup))); } IndexTuple leftHightKey = UBTreePCRGetIndexTuple(leftpage, P_HIKEY); @@ -2929,7 +2931,7 @@ void LogSplit(Buffer buf, Buffer rbuf, Buffer sbuf, Buffer leftCbuf, OffsetNumbe if (isroot) { xlflag = newItemOnLeft ? XLOG_UBTREE3_SPLIT_L_ROOT : XLOG_UBTREE3_SPLIT_R_ROOT; } else { - xlflag = newItemOnLeft ? XLOG_UBTREE3_SPLIT_L : XLOG_UBTREE_SPLIT_R; + xlflag = newItemOnLeft ? XLOG_UBTREE3_SPLIT_L : XLOG_UBTREE3_SPLIT_R; } recptr = XLogInsert(RM_UBTREE3_ID, xlflag); diff --git a/src/gausskernel/storage/access/ubtreepcr/ubtpcrpage.cpp b/src/gausskernel/storage/access/ubtreepcr/ubtpcrpage.cpp index 15bf27aeaec3ed12f3455e32e6fac898b898bca6..29dfee56b156c79ae725c85d95ab76d17e805bb3 100644 --- a/src/gausskernel/storage/access/ubtreepcr/ubtpcrpage.cpp +++ b/src/gausskernel/storage/access/ubtreepcr/ubtpcrpage.cpp @@ -244,21 +244,23 @@ Buffer UBTreePCRGetRoot(Relation rel, int access) /* XLOG stuff */ if (RelationNeedsWAL(rel)) { - xl_btree_newroot xlrec; - xl_btree_metadata_old md; - XLogRecPtr recptr; XLogBeginInsert(); XLogRegisterBuffer(0, rootbuf, REGBUF_WILL_INIT); XLogRegisterBuffer(2, metabuf, REGBUF_WILL_INIT); + xl_btree_metadata md; md.root = rootblkno; md.level = 0; md.fastroot = rootblkno; md.fastlevel = 0; - XLogRegisterBufData(2, (char *)&md, sizeof(xl_btree_metadata_old)); + md.version = metad->btm_version; + XLogRegisterBufData(2, (char *)&md, sizeof(xl_btree_metadata)); + + xl_btree_newroot xlrec; xlrec.rootblk = rootblkno; xlrec.level = 0; XLogRegisterData((char *)&xlrec, SizeOfBtreeNewroot); - recptr = XLogInsert(RM_UBTREE3_ID, XLOG_UBTREE3_NEW_ROOT); + + XLogRecPtr recptr = XLogInsert(RM_UBTREE3_ID, XLOG_UBTREE3_NEW_ROOT); PageSetLSN(rootpage, recptr); PageSetLSN(metapg, recptr); } diff --git a/src/gausskernel/storage/access/ubtreepcr/ubtpcrrollback.cpp b/src/gausskernel/storage/access/ubtreepcr/ubtpcrrollback.cpp index 8e2666da2191ff3d621d5b965513c6e42d1f3b5f..4207b8169fbc42d856c5eb33ae70c933b7884ba9 100644 --- a/src/gausskernel/storage/access/ubtreepcr/ubtpcrrollback.cpp +++ b/src/gausskernel/storage/access/ubtreepcr/ubtpcrrollback.cpp @@ -670,7 +670,7 @@ int UBTreePCRRollback(URecVector *urecvec, int startIdx, int endIdx, Transaction XLogRegisterData((char *)&xlrec, sizeOfUbtree3RollbackTxn); XLogRegisterData((char *)&redoRollbackAction.td, sizeof(UBTreeTDData)); if (xlrec.n_rollback > 0) { - XLogRegisterData((char *)&redoRollbackAction.rollback_items.items, + XLogRegisterData((char *)redoRollbackAction.rollback_items.items, sizeof(UBTreeRedoRollbackItemData) * xlrec.n_rollback); } XLogRegisterBuffer(0, buffer, REGBUF_STANDARD); @@ -758,7 +758,7 @@ void ExecuteUndoActionsForUBTreePage(Relation rel, Buffer buffer, uint8 tdid) XLogRegisterData((char *)&xlrec, sizeOfUbtree3RollbackTxn); XLogRegisterData((char *)&redoRollbackAction.td, sizeof(UBTreeTDData)); if (xlrec.n_rollback > 0) { - XLogRegisterData((char *)&redoRollbackAction.rollback_items.items, + XLogRegisterData((char *)redoRollbackAction.rollback_items.items, sizeof(UBTreeRedoRollbackItemData) * xlrec.n_rollback); } diff --git a/src/gausskernel/storage/access/ubtreepcr/ubtpcrsearch.cpp b/src/gausskernel/storage/access/ubtreepcr/ubtpcrsearch.cpp index 9cd5226d9c7d7927bd21099cd517b74229e9f022..27c89e89f310cd187983de3817c41ed892a37f1e 100644 --- a/src/gausskernel/storage/access/ubtreepcr/ubtpcrsearch.cpp +++ b/src/gausskernel/storage/access/ubtreepcr/ubtpcrsearch.cpp @@ -50,7 +50,7 @@ static void UBTreePCRSaveItem(IndexScanDesc scan, int itemIndex, Page page, Offs const IndexTuple itup, Oid partOid); static bool UBTreePCRStepPage(IndexScanDesc scan, ScanDirection dir); static bool UBTreePCREndPoint(IndexScanDesc scan, ScanDirection dir); -static void BuildCRPage(IndexScanDesc scan, Page crPage, Buffer baseBuffer, CommandId *page_cid); +static void BuildCRPage(IndexScanDesc scan, Page crPage, Buffer baseBuffer); const uint16 INVALID_TUPLE_OFFSET = (uint16)0xa5a5; @@ -1687,7 +1687,6 @@ static bool UBTreePCRReadPage(IndexScanDesc scan, ScanDirection dir, OffsetNumbe Snapshot snapshot = scan->xs_snapshot; Page crPage = NULL; - CRBufferDesc* desc = NULL; bool canUsePCRScanMode = IsPCRScanModeSupported(snapshot); /* @@ -1699,8 +1698,6 @@ static bool UBTreePCRReadPage(IndexScanDesc scan, ScanDirection dir, OffsetNumbe ((snapshot->satisfies == SNAPSHOT_MVCC || snapshot->satisfies == SNAPSHOT_VERSION_MVCC) && TransactionIdFollowsOrEquals(opaque->last_delete_xid, snapshot->xmin)); - Page localPage = (Page)palloc(BLCKSZ); - bool useLocalPage = false; uint32 checkNum = 0; OffsetNumber checkVisibleOffs[MaxIndexTuplesPerPage] = {0}; OffsetNumber originOffnum = offnum; @@ -1717,43 +1714,14 @@ choose_scan_mode: ereport(DEBUG3, (errmodule(MOD_PCR), (errmsg("ReadCRBuffer begin (%u, %d), " "querycsn %ld, querycid %d", scan->indexRelation->rd_node.relNode, blockNum, snapshot->snapshotcsn, snapshot->curcid)))); - desc = ReadCRBuffer(scan->indexRelation, BufferGetBlockNumber(so->currPos.buf), - snapshot->snapshotcsn, snapshot->curcid); - if (desc != NULL) { - ereport(DEBUG3, (errmodule(MOD_PCR), (errmsg("ReadCRBuffer found (%u, %d) buf_id %d, " - "csn %ld, cid %d, rsid %ld", scan->indexRelation->rd_node.relNode, - blockNum, desc->buf_id, desc->csn, desc->cid, desc->rsid)))); - crPage = CRBufferGetPage(desc->buf_id); - } else if (mustUsePCRScanMode) { - ereport(DEBUG5, (errmodule(MOD_PCR), (errmsg("PCR read page from cr pool notfound")))); - errno_t rc = memcpy_sp(localPage, BLCKSZ, (Page)BufferGetPage(so->currPos.buf), BLCKSZ); - CommandId page_cid; - BuildCRPage(scan, localPage, so->currPos.buf, &page_cid); - page_cid = page_cid == InvalidCommandId ? snapshot->curcid : page_cid; - UBTPCRPageOpaque opaque = (UBTPCRPageOpaque)PageGetSpecialPointer(page); - - if (!((opaque->btpo_flags & BTP_DELETED) || (opaque->btpo_flags & BTP_HALF_DEAD) || - (opaque->btpo_flags & BTP_VACUUM_DELETING) || (opaque->btpo_flags & BTP_INCOMPLETE_SPLIT) || - (opaque->btpo_flags & BTP_SPLIT_END))) { - ereport(DEBUG3, (errmodule(MOD_PCR), (errmsg("AllocCRBuffer (%u, %d) page_csn %ld, page_cid %d", - scan->indexRelation->rd_node.relNode, blockNum, - snapshot->snapshotcsn, page_cid)))); - desc = AllocCRBuffer(scan->indexRelation, MAIN_FORKNUM, BufferGetBlockNumber(so->currPos.buf), - snapshot->snapshotcsn, page_cid); - crPage = CRBufferGetPage(desc->buf_id); - rc = memcpy_sp(crPage, BLCKSZ, (Page)localPage, BLCKSZ); - uint64 buf_state = pg_atomic_read_u64(&desc->state); - UnlockCRBufHdr(desc, buf_state); - } else { - crPage = localPage; - } + if (mustUsePCRScanMode) { + crPage = (Page)palloc(BLCKSZ); + BuildCRPage(scan, crPage, so->currPos.buf); so->scanMode = PCR_SCAN_MODE; - page = crPage; - maxoff = UBTreePCRPageGetMaxOffsetNumber(page); } else { so->scanMode = PBRCR_SCAN_MODE; } - if (desc != NULL) { + if (crPage != NULL) { so->scanMode = PCR_SCAN_MODE; page = crPage; maxoff = UBTreePCRPageGetMaxOffsetNumber(page); @@ -1890,13 +1858,9 @@ choose_scan_mode: so->currPos.firstItem = itemIndex; } } - if (desc != NULL) { - ReleaseCRBuffer(desc->buf_id); - ereport(DEBUG3, (errmodule(MOD_PCR), (errmsg("release cr buffer (%u, %d), buf_id %d, csn %ld, cid %d, " - "rsid %ld", scan->indexRelation->rd_node.relNode, BufferGetBlockNumber(so->currPos.buf), - desc->buf_id, desc->csn, desc->cid, desc->rsid)))); + if (crPage != NULL) { + pfree(crPage); } - pfree(localPage); return (so->currPos.firstItem <= so->currPos.lastItem); } @@ -2520,7 +2484,7 @@ int TDCompare(Datum a, Datum b, void *arg) } } -static void BuildCRPage(IndexScanDesc scan, Page crPage, Buffer baseBuffer, CommandId *page_cid) +static void BuildCRPage(IndexScanDesc scan, Page crPage, Buffer baseBuffer) { /* copy base page to cr page and unlock base page */ Page basePage = BufferGetPage(baseBuffer); @@ -2531,7 +2495,6 @@ static void BuildCRPage(IndexScanDesc scan, Page crPage, Buffer baseBuffer, Comm Relation rel = scan->indexRelation; Snapshot snapshot = scan->xs_snapshot; - *page_cid = InvalidCommandId; if (!IsPCRScanModeSupported(snapshot)) { ereport(ERROR, (errmsg("Unsupported snapshot type %u when contruct cr page rnode[%u,%u,%u], blkno:%u", @@ -2606,7 +2569,6 @@ loop: RollbackCRPage(scan, crPage, tdid, &cur_page_cid, cid); if (xid == td->xactid) { Assert(cid > 0); - *page_cid = cur_page_cid; continue; } prevInprogress = true; diff --git a/src/gausskernel/storage/access/ubtreepcr/ubtpcrtd.cpp b/src/gausskernel/storage/access/ubtreepcr/ubtpcrtd.cpp index 8adf58d25b9e7f7a89bdcd3e7a73d396218704af..7b92cefcd6ed7f940f35e889c36a9b4bf5065556 100644 --- a/src/gausskernel/storage/access/ubtreepcr/ubtpcrtd.cpp +++ b/src/gausskernel/storage/access/ubtreepcr/ubtpcrtd.cpp @@ -180,12 +180,13 @@ uint8 UBTreePageFreezeTDSlots(Relation relation, Buffer buf, TransactionId* minX XLogBeginInsert(); xlrec.nFrozen = nFrozenSlots; + xlrec.latestFrozenXid = latestfxid; XLogRegisterData((char*)&xlrec, SizeOfUbtree3FreezeTDSlot); /* * We need the frozen slots information when WAL needs to * be applied on the page.. */ - XLogRegisterData((char *)frozenSlots, nFrozenSlots * sizeof(int)); + XLogRegisterData((char *)frozenSlots, nFrozenSlots * sizeof(uint8)); XLogRegisterBuffer(0, buf, REGBUF_STANDARD); recptr = XLogInsert(RM_UBTREE3_ID, XLOG_UBTREE3_FREEZE_TD_SLOT); @@ -263,8 +264,8 @@ uint8 UBTreePageFreezeTDSlots(Relation relation, Buffer buf, TransactionId* minX if (RelationNeedsWAL(relation)) { XLogBeginInsert(); - XLogRegisterData((char *)&nCompletedXactSlots, sizeof(uint16)); - XLogRegisterData((char *)completedXactSlots, nCompletedXactSlots * sizeof(int)); + XLogRegisterData((char *)&nCompletedXactSlots, sizeof(uint8)); + XLogRegisterData((char *)completedXactSlots, nCompletedXactSlots * sizeof(uint8)); XLogRegisterBuffer(0, buf, REGBUF_STANDARD); @@ -387,7 +388,7 @@ uint8 UBTreeExtendTDSlots(Relation relation, Buffer buf) XLogRegisterData((char *) &xlrec, SizeOfUbtree3ExtendTDSlot); XLogRegisterBuffer(0, buf, REGBUF_STANDARD); - recptr = XLogInsert(RM_UHEAP2_ID, XLOG_UBTREE3_EXTEND_TD_SLOTS); + recptr = XLogInsert(RM_UBTREE3_ID, XLOG_UBTREE3_EXTEND_TD_SLOTS); PageSetLSN(page, recptr); } diff --git a/src/gausskernel/storage/access/ubtreepcr/ubtpcrundo.cpp b/src/gausskernel/storage/access/ubtreepcr/ubtpcrundo.cpp index 9e987d772c48f65dbbece8f61938741fe1958161..777d74f43fdaa4b6ecb5edef78daf948a4b4f53d 100644 --- a/src/gausskernel/storage/access/ubtreepcr/ubtpcrundo.cpp +++ b/src/gausskernel/storage/access/ubtreepcr/ubtpcrundo.cpp @@ -29,7 +29,7 @@ UndoRecPtr UBTreePCRPrepareUndoInsert(Oid relOid, Oid partitionOid, Oid relfilenode, Oid tablespace, UndoPersistence persistence, TransactionId xid, CommandId cid, UndoRecPtr prevurpInOneBlk, UndoRecPtr prevurpInOneXact, BlockNumber blk, XlUndoHeader *xlundohdr, undo::XlogUndoMeta *xlundometa, - OffsetNumber offset, Buffer buf, bool selfInsert, UBTreeUndoInfo undoinfo, IndexTuple itup) + OffsetNumber offset, Buffer buf, TransactionId oldXid, UBTreeUndoInfo undoinfo, IndexTuple itup) { UndoRecord *urec = (*t_thrd.ustore_cxt.urecvec)[0]; Assert(tablespace != InvalidOid); @@ -56,12 +56,8 @@ UndoRecPtr UBTreePCRPrepareUndoInsert(Oid relOid, Oid partitionOid, Oid relfilen } } /* Tell Undo chain traversal this record does not have any older version */ - if (selfInsert) { - urec->SetOldXactId(xid); - } else { - urec->SetOldXactId(FrozenTransactionId); - } - + urec->SetOldXactId(oldXid); + MemoryContext old_cxt = MemoryContextSwitchTo(urec->mem_context()); initStringInfo(urec->Rawdata()); MemoryContextSwitchTo(old_cxt); diff --git a/src/gausskernel/storage/access/ustore/knl_uundovec.cpp b/src/gausskernel/storage/access/ustore/knl_uundovec.cpp index 9e221aee85214296392622e91177f6a43b003080..636cd5f47a280991aa4a3192bd391a30e2e3befe 100644 --- a/src/gausskernel/storage/access/ustore/knl_uundovec.cpp +++ b/src/gausskernel/storage/access/ustore/knl_uundovec.cpp @@ -556,7 +556,7 @@ static bool CheckLastRecordSize(UndoRecordSize lastRecordSize, undo::XlogUndoMet if (t_thrd.xlog_cxt.InRecovery && (lastRecordSize != xlundometa->lastRecordSize)) { ereport(PANIC, (errmsg(UNDOFORMAT("last record size %u != xlog last record size %u," - "xlog info %d, slotPtr %lu,dbid %u."), + "xlog info %d, slotPtr %lu,dbid %u."), lastRecordSize, xlundometa->lastRecordSize, xlundometa->info, xlundometa->slotPtr, xlundometa->dbid))); return false; diff --git a/src/include/access/nbtree.h b/src/include/access/nbtree.h index a480499b92854ce35cb7f0ecc20ad43d71f521fc..eaba2380902f6431d9e1db58b4de8a4b59018446 100644 --- a/src/include/access/nbtree.h +++ b/src/include/access/nbtree.h @@ -392,7 +392,6 @@ enum { BTREE_UNLINK_PAGE_META_NUM, }; - enum { BTREE_NEWROOT_ORIG_BLOCK_NUM = 0, BTREE_NEWROOT_LEFT_BLOCK_NUM, diff --git a/src/include/access/ubtreepcr.h b/src/include/access/ubtreepcr.h index 01f21d302053e2637f3350a5ab22ef70751b5b72..9b67861268e55732c6b9728e47728479916a342f 100644 --- a/src/include/access/ubtreepcr.h +++ b/src/include/access/ubtreepcr.h @@ -367,7 +367,7 @@ extern void RollbackCRPage(IndexScanDesc scan, Page crPage, uint8 tdid, extern UndoRecPtr UBTreePCRPrepareUndoInsert(Oid relOid, Oid partitionOid, Oid relfilenode, Oid tablespace, UndoPersistence persistence, TransactionId xid, CommandId cid, UndoRecPtr prevurpInOneBlk, UndoRecPtr prevurpInOneXact, BlockNumber blk, XlUndoHeader *xlundohdr, undo::XlogUndoMeta *xlundometa, - OffsetNumber offset, Buffer buf, bool selfInsert, UBTreeUndoInfo undoinfo, IndexTuple itup); + OffsetNumber offset, Buffer buf, TransactionId oldXid, UBTreeUndoInfo undoinfo, IndexTuple itup); extern int UBTreePCRRollback(URecVector *urecvec, int startIdx, int endIdx, TransactionId xid, Oid reloid, Oid partitionoid, BlockNumber blkno, bool isFullChain, int preRetCode, Oid *preReloid, Oid *prePartitionoid); /* diff --git a/src/include/access/xlogproc.h b/src/include/access/xlogproc.h index eab504179b346c240c8551bea49a08b248f89970..ba7a0fbf91ea260ddd3a6ddfe597eecfbab4a726 100755 --- a/src/include/access/xlogproc.h +++ b/src/include/access/xlogproc.h @@ -1073,9 +1073,13 @@ template void UBTreeXlogClearIncompleteSplit(RedoBufferInfo* bu /* UBT PCR */ extern void UBTree3XlogPrunePageOperatorPage(RedoBufferInfo* buffer, void* recorddata, const Oid rdid); extern void UBTree3XlogDeleteOperatorPage(RedoBufferInfo* buffer, void* recorddata, UndoRecPtr urecptr); +extern void UBTree3XlogFreezeTdOperatorPage(RedoBufferInfo* buffer, void* recorddata); extern void UBTree3XlogReuseTdOperatorPage(RedoBufferInfo* buffer, void* recorddata); +extern void UBTree3XlogExtendTdOperatorPage(RedoBufferInfo* buffer, void* recorddata); extern void UBTree3RestoreMetaOperatorPage(RedoBufferInfo *metabuf, void *recorddata, Size datalen); extern void UBTree3XlogRollbackTxnOperatorPage(RedoBufferInfo* buffer, void* recorddata); +extern void UBTree3XlogSplitLeftPage(RedoBufferInfo *bufferinfo, void *recorddata, bool onleft, + void *blkdata, Size datalen); extern void UBTree4XlogUnlinkPageOperatorRightpage(RedoBufferInfo *rbuf, void *recorddata); extern void UBTree4XlogUnlinkPageOperatorLeftpage(RedoBufferInfo *lbuf, void *recorddata); extern void UBTree4XlogUnlinkPageOperatorCurpage(RedoBufferInfo *buf, void *recorddata); @@ -1320,8 +1324,6 @@ void segpage_redo_new_page_for_standby_read(XLogBlockSegNewPage *block_data_rec, void redo_target_page(const BufferTag &old_buf_tag, const BufferTag &new_buf_tag, StandbyReadLsnInfoArray *lsn_info, Buffer base_page_buf); void MarkSegPageRedoChildPageDirty(RedoBufferInfo *bufferinfo); -extern void BTree3XlogSplitLeftPage(RedoBufferInfo *bufferinfo, void *recorddata, bool onleft, - void *blkdata, Size datalen); // shared-storage XLogRedoAction SSCheckInitPageXLog(XLogReaderState *record, uint8 block_id, RedoBufferInfo *redo_buf);