diff --git a/src/backend/access/gin/README b/src/backend/access/gin/README index 67159d8..6ebee1b 100644 --- a/src/backend/access/gin/README +++ b/src/backend/access/gin/README @@ -145,6 +145,7 @@ none appear in the key entry itself. The separate pages are called a Note that in either case, the ItemPointers associated with a key can easily be read out in sorted order; this is relied on by the scan algorithms. +FIXME: Update above paragraph! * The index tuple header fields of a leaf key entry are abused as follows: diff --git a/src/backend/access/gin/gindatapage.c b/src/backend/access/gin/gindatapage.c index f017de0..a250bfc 100644 --- a/src/backend/access/gin/gindatapage.c +++ b/src/backend/access/gin/gindatapage.c @@ -18,6 +18,97 @@ #include "utils/rel.h" /* + * Write item pointer into leaf data page using varbyte encoding. Since + * BlockNumber is stored in incremental manner we also need a previous item + * pointer. + */ +char * +ginDataPageLeafWriteItemPointer(char *ptr, ItemPointer iptr, ItemPointer prev) +{ + uint32 blockNumberIncr; + uint16 offset; + uint8 v; + + blockNumberIncr = iptr->ip_blkid.bi_lo + (iptr->ip_blkid.bi_hi << 16) - + (prev->ip_blkid.bi_lo + (prev->ip_blkid.bi_hi << 16)); + for (;;) + { + if (blockNumberIncr < HIGHBIT) + { + v = (uint8) blockNumberIncr; + *ptr = v; + ptr++; + break; + } + else + { + v = ((uint8) blockNumberIncr) | HIGHBIT; + *ptr = v; + ptr++; + blockNumberIncr >>= 7; + } + } + + offset = iptr->ip_posid; + for (;;) + { + if (offset < HIGHBIT) + { + v = (uint8) offset; + *ptr = v; + ptr++; + break; + } + else + { + v = ((uint8) offset) | HIGHBIT; + *ptr = v; + ptr++; + offset >>= 7; + } + } + + return ptr; +} + +/* + * Calculate size of incremental varbyte encoding of item pointer. + */ +static int +ginDataPageLeafGetItemPointerSize(ItemPointer iptr, ItemPointer prev) +{ + uint32 blockNumberIncr; + uint16 offset; + int size = 0; + + blockNumberIncr = iptr->ip_blkid.bi_lo + (iptr->ip_blkid.bi_hi << 16) - + (prev->ip_blkid.bi_lo + (prev->ip_blkid.bi_hi << 16)); + do + { + size++; + blockNumberIncr >>= 7; + } while (blockNumberIncr > 0); + + offset = iptr->ip_posid; + do + { + size++; + offset >>= 7; + } while (offset > 0); + + return size; +} + +/* + * Returns size of item pointers if leaf data page after inserting another one. + */ +Size +ginCheckPlaceToDataPageLeaf(ItemPointer iptr, ItemPointer prev, Size size) +{ + return size + ginDataPageLeafGetItemPointerSize(iptr, prev); +} + +/* * Merge two ordered arrays of itempointers, eliminating any duplicates. * Returns the number of items in the result. * Caller is responsible that there is enough space at *dst. @@ -91,12 +182,12 @@ dataLocateItem(GinBtree btree, GinBtreeStack *stack) if (btree->fullScan) { stack->off = FirstOffsetNumber; - stack->predictNumber *= GinPageGetOpaque(page)->maxoff; + stack->predictNumber *= GinPageGetOpaque(page)->u.maxoff; return btree->getLeftMostPage(btree, page); } low = FirstOffsetNumber; - maxoff = high = GinPageGetOpaque(page)->maxoff; + maxoff = high = GinPageGetOpaque(page)->u.maxoff; Assert(high >= low); high++; @@ -105,7 +196,7 @@ dataLocateItem(GinBtree btree, GinBtreeStack *stack) { OffsetNumber mid = low + ((high - low) / 2); - pitem = (PostingItem *) GinDataPageGetItem(page, mid); + pitem = GinDataPageGetItem(page, mid); if (mid == maxoff) { @@ -117,7 +208,7 @@ dataLocateItem(GinBtree btree, GinBtreeStack *stack) } else { - pitem = (PostingItem *) GinDataPageGetItem(page, mid); + pitem = GinDataPageGetItem(page, mid); result = ginCompareItemPointers(btree->items + btree->curitem, &(pitem->key)); } @@ -135,11 +226,82 @@ dataLocateItem(GinBtree btree, GinBtreeStack *stack) Assert(high >= FirstOffsetNumber && high <= maxoff); stack->off = high; - pitem = (PostingItem *) GinDataPageGetItem(page, high); + pitem = GinDataPageGetItem(page, high); + return PostingItemGetBlockNumber(pitem); } /* + * Find item pointer in leaf data page. Returns true if given item pointer is + * found and false if it's not. Sets offset and iptrOut to last item pointer + * which is less than given one. Sets ptrOut ahead that item pointer. + */ +static bool +findInLeafPage(GinBtree btree, Page page, OffsetNumber *offset, + ItemPointerData *iptrOut, Pointer *ptrOut) +{ + Pointer ptr = GinDataPageGetData(page); + int i; + OffsetNumber off; + ItemPointerData iptr = {{0,0},0}; + int cmp; + Pointer endPtr; + bool result = false; + + endPtr = page + GinPageGetOpaque(page)->u.endoffset; + + /* + * At first, search index at the end of page. As the result we narrow + * [first, maxoff] range. + */ + off = FirstOffsetNumber; + for (i = 0; i < GinDataLeafIndexCount; i++) + { + GinDataLeafItemIndex *index = &GinPageGetIndexes(page)[i]; + if (index->offsetNumber == InvalidOffsetNumber) + break; + + cmp = ginCompareItemPointers(&index->iptr, btree->items + btree->curitem); + if (cmp < 0) + { + ptr = GinDataPageGetData(page) + index->pageOffset; + iptr = index->iptr; + off = index->offsetNumber; + } + else + { + endPtr = page + index->pageOffset; + break; + } + } + + /* Search page in [first, maxoff] range found by page index */ + while (ptr < endPtr) + { + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + + cmp = ginCompareItemPointers(btree->items + btree->curitem, &iptr); + if (cmp == 0) + { + result = true; + break; + } + if (cmp < 0) + { + result = false; + break; + } + off++; + } + + *ptrOut = ptr; + *iptrOut = iptr; + *offset = off; + return result; +} + + +/* * Searches correct position for value on leaf page. * Page should be correctly chosen. * Returns true if value found on page. @@ -148,9 +310,8 @@ static bool dataLocateLeafItem(GinBtree btree, GinBtreeStack *stack) { Page page = BufferGetPage(stack->buffer); - OffsetNumber low, - high; - int result; + ItemPointerData iptr; + Pointer ptr; Assert(GinPageIsLeaf(page)); Assert(GinPageIsData(page)); @@ -161,36 +322,7 @@ dataLocateLeafItem(GinBtree btree, GinBtreeStack *stack) return TRUE; } - low = FirstOffsetNumber; - high = GinPageGetOpaque(page)->maxoff; - - if (high < low) - { - stack->off = FirstOffsetNumber; - return false; - } - - high++; - - while (high > low) - { - OffsetNumber mid = low + ((high - low) / 2); - - result = ginCompareItemPointers(btree->items + btree->curitem, (ItemPointer) GinDataPageGetItem(page, mid)); - - if (result == 0) - { - stack->off = mid; - return true; - } - else if (result > 0) - low = mid + 1; - else - high = mid; - } - - stack->off = high; - return false; + return findInLeafPage(btree, page, &stack->off, &iptr, &ptr); } /* @@ -201,7 +333,7 @@ static OffsetNumber dataFindChildPtr(GinBtree btree, Page page, BlockNumber blkno, OffsetNumber storedOff) { OffsetNumber i, - maxoff = GinPageGetOpaque(page)->maxoff; + maxoff = GinPageGetOpaque(page)->u.maxoff; PostingItem *pitem; Assert(!GinPageIsLeaf(page)); @@ -210,7 +342,7 @@ dataFindChildPtr(GinBtree btree, Page page, BlockNumber blkno, OffsetNumber stor /* if page isn't changed, we return storedOff */ if (storedOff >= FirstOffsetNumber && storedOff <= maxoff) { - pitem = (PostingItem *) GinDataPageGetItem(page, storedOff); + pitem = GinDataPageGetItem(page, storedOff); if (PostingItemGetBlockNumber(pitem) == blkno) return storedOff; @@ -220,7 +352,7 @@ dataFindChildPtr(GinBtree btree, Page page, BlockNumber blkno, OffsetNumber stor */ for (i = storedOff + 1; i <= maxoff; i++) { - pitem = (PostingItem *) GinDataPageGetItem(page, i); + pitem = GinDataPageGetItem(page, i); if (PostingItemGetBlockNumber(pitem) == blkno) return i; } @@ -231,7 +363,7 @@ dataFindChildPtr(GinBtree btree, Page page, BlockNumber blkno, OffsetNumber stor /* last chance */ for (i = FirstOffsetNumber; i <= maxoff; i++) { - pitem = (PostingItem *) GinDataPageGetItem(page, i); + pitem = GinDataPageGetItem(page, i); if (PostingItemGetBlockNumber(pitem) == blkno) return i; } @@ -249,37 +381,38 @@ dataGetLeftMostPage(GinBtree btree, Page page) Assert(!GinPageIsLeaf(page)); Assert(GinPageIsData(page)); - Assert(GinPageGetOpaque(page)->maxoff >= FirstOffsetNumber); + Assert(GinPageGetOpaque(page)->u.maxoff >= FirstOffsetNumber); - pitem = (PostingItem *) GinDataPageGetItem(page, FirstOffsetNumber); + pitem = GinDataPageGetItem(page, FirstOffsetNumber); return PostingItemGetBlockNumber(pitem); } /* - * add ItemPointer or PostingItem to page. data should point to + * add PostingItem to page. data should point to * correct value! depending on leaf or non-leaf page */ void -GinDataPageAddItem(Page page, void *data, OffsetNumber offset) +GinPageAddPostingItem(Page page, PostingItem *data, OffsetNumber offset) { - OffsetNumber maxoff = GinPageGetOpaque(page)->maxoff; - char *ptr; + OffsetNumber maxoff = GinPageGetOpaque(page)->u.maxoff; + PostingItem *ptr; + + Assert(!GinPageIsLeaf(page)); if (offset == InvalidOffsetNumber) - { ptr = GinDataPageGetItem(page, maxoff + 1); - } else { + Assert(offset <= maxoff + 1); ptr = GinDataPageGetItem(page, offset); - if (maxoff + 1 - offset != 0) - memmove(ptr + GinSizeOfDataPageItem(page), + if (offset != maxoff + 1) + memmove(GinDataPageGetItem(page, offset +1), ptr, - (maxoff - offset + 1) * GinSizeOfDataPageItem(page)); + (maxoff - offset + 1) * sizeof(PostingItem)); } - memcpy(ptr, data, GinSizeOfDataPageItem(page)); + memcpy(ptr, data, sizeof(PostingItem)); - GinPageGetOpaque(page)->maxoff++; + GinPageGetOpaque(page)->u.maxoff++; } /* @@ -288,16 +421,17 @@ GinDataPageAddItem(Page page, void *data, OffsetNumber offset) void GinPageDeletePostingItem(Page page, OffsetNumber offset) { - OffsetNumber maxoff = GinPageGetOpaque(page)->maxoff; + OffsetNumber maxoff = GinPageGetOpaque(page)->u.maxoff; Assert(!GinPageIsLeaf(page)); Assert(offset >= FirstOffsetNumber && offset <= maxoff); if (offset != maxoff) - memmove(GinDataPageGetItem(page, offset), GinDataPageGetItem(page, offset + 1), + memmove(GinDataPageGetItem(page, offset), + GinDataPageGetItem(page, offset + 1), sizeof(PostingItem) * (maxoff - offset)); - GinPageGetOpaque(page)->maxoff--; + GinPageGetOpaque(page)->u.maxoff--; } /* @@ -314,15 +448,26 @@ dataIsEnoughSpace(GinBtree btree, Buffer buf, OffsetNumber off) if (GinPageIsLeaf(page)) { - if (GinPageRightMost(page) && off > GinPageGetOpaque(page)->maxoff) + int j; + ItemPointerData iptr = {{0,0},0}; + Size size = 0; + + /* + * Calculate additional size using worst case assumption: varbyte + * encoding from zero item pointer. + */ + for (j = btree->curitem; j < btree->nitem; j++) { - if ((btree->nitem - btree->curitem) * sizeof(ItemPointerData) <= GinDataPageGetFreeSpace(page)) - return true; + size = ginCheckPlaceToDataPageLeaf(&btree->items[j], + (j == btree->curitem) ? (&iptr) : &btree->items[j - 1], + size); } - else if (sizeof(ItemPointerData) <= GinDataPageGetFreeSpace(page)) + + if (GinLeafDataPageGetFreeSpace(page) >= size) return true; + } - else if (sizeof(PostingItem) <= GinDataPageGetFreeSpace(page)) + else if (sizeof(PostingItem) <= GinNonLeafDataPageGetFreeSpace(page)) return true; return false; @@ -361,12 +506,12 @@ static void dataPlaceToPage(GinBtree btree, Buffer buf, OffsetNumber off, XLogRecData **prdata) { Page page = BufferGetPage(buf); - int sizeofitem = GinSizeOfDataPageItem(page); int cnt = 0; /* these must be static so they can be returned to caller */ static XLogRecData rdata[3]; static ginxlogInsert data; + static char insertData[BLCKSZ]; *prdata = rdata; Assert(GinPageIsData(page)); @@ -376,7 +521,7 @@ dataPlaceToPage(GinBtree btree, Buffer buf, OffsetNumber off, XLogRecData **prda data.node = btree->index->rd_node; data.blkno = BufferGetBlockNumber(buf); data.offset = off; - data.nitem = 1; + data.nitem = 0; data.isDelete = FALSE; data.isData = TRUE; data.isLeaf = GinPageIsLeaf(page) ? TRUE : FALSE; @@ -404,35 +549,343 @@ dataPlaceToPage(GinBtree btree, Buffer buf, OffsetNumber off, XLogRecData **prda rdata[cnt].next = &rdata[cnt + 1]; cnt++; - rdata[cnt].buffer = InvalidBuffer; - rdata[cnt].data = (GinPageIsLeaf(page)) ? ((char *) (btree->items + btree->curitem)) : ((char *) &(btree->pitem)); - rdata[cnt].len = sizeofitem; - rdata[cnt].next = NULL; - if (GinPageIsLeaf(page)) { - if (GinPageRightMost(page) && off > GinPageGetOpaque(page)->maxoff) + int i = 0, j, max_j; + Pointer ptr = GinDataPageGetData(page), next_ptr, insertStart; + ItemPointerData iptr = {{0,0},0}, next_iptr; + char pageCopy[BLCKSZ]; + int copySize = 0; + char *endPtr = page + GinPageGetOpaque(page)->u.endoffset; + bool append = true; + + /* + * We're going to prevent var-byte re-encoding of whole page. + * Find position in page using page indexes. + */ + findInLeafPage(btree, page, &off, &iptr, &ptr); + + Assert(GinDataPageFreeSpacePre(page,ptr) >= 0); + + if (ptr < endPtr) { - /* usually, create index... */ - uint32 savedPos = btree->curitem; + /* + * Read next item pointer: we'll have to re-encode it. Copy + * previous part of page + */ + next_iptr = iptr; + next_ptr = ginDataPageLeafReadItemPointer(ptr, &next_iptr); + copySize = GinDataPageSize - GinLeafDataPageGetFreeSpace(page) - + (next_ptr - GinDataPageGetData(page)); + memcpy(pageCopy, next_ptr, copySize); + append = false; + } - while (btree->curitem < btree->nitem) + /* Check how many items we're going to add */ + max_j = btree->curitem + btree->nitem; + + /* Place items to the page while we have enough of space */ + memcpy(insertData, &iptr, sizeof(ItemPointerData)); + insertStart = ptr; + i = 0; + for (j = btree->curitem; j < max_j; j++) + { + Pointer ptr2; + + ptr2 = page + ginCheckPlaceToDataPageLeaf(&btree->items[j], + &iptr, ptr - page); + + if (GinDataPageFreeSpacePre(page, ptr2) < 0) + break; + + ptr = ginDataPageLeafWriteItemPointer(ptr, &btree->items[j], &iptr); + Assert(GinDataPageFreeSpacePre(page,ptr) >= 0); + + iptr = btree->items[j]; + btree->curitem++; + data.nitem++; + i++; + } + Assert(i > 0); + + /* Put WAL data */ + memcpy(insertData + sizeof(ItemPointerData), insertStart, + ptr - insertStart); + rdata[cnt].buffer = InvalidBuffer; + rdata[cnt].data = insertData; + rdata[cnt].len = sizeof(ItemPointerData) + (ptr - insertStart); + rdata[cnt].next = NULL; + + /* Place rest of the page back */ + if (!append) + { + ptr = ginDataPageLeafWriteItemPointer(ptr, &next_iptr, &iptr); + Assert(GinDataPageFreeSpacePre(page,ptr) >= 0); + memcpy(ptr, pageCopy, copySize); + ptr += copySize; + } + + GinPageGetOpaque(page)->u.endoffset = ptr - page; + + if (GinDataPageFreeSpacePre(page,ptr) < 0) + elog(ERROR, "not enough space in leaf page!"); + + /* Update indexes in the end of page */ + updateItemIndexes(page, btree->ginstate); + } + else + { + rdata[cnt].buffer = InvalidBuffer; + rdata[cnt].data = (char *) &(btree->pitem); + rdata[cnt].len = sizeof(PostingItem); + rdata[cnt].next = NULL; + data.nitem = 1; + + GinPageAddPostingItem(page, &(btree->pitem), off); + } +} + +/* Macro for leaf data page split: switch to right page if needed. */ +#define CHECK_SWITCH_TO_RPAGE \ + do { \ + if (ptr - GinDataPageGetData(page) > \ + totalsize / 2 && page == lpage) \ + { \ + maxLeftIptr = iptr; \ + prevIptr.ip_blkid.bi_hi = 0; \ + prevIptr.ip_blkid.bi_lo = 0; \ + prevIptr.ip_posid = 0; \ + GinPageGetOpaque(lpage)->u.endoffset = ptr - page; \ + page = rpage; \ + ptr = GinDataPageGetData(rpage); \ + separator = j; \ + j = FirstOffsetNumber; \ + } \ + else \ + { \ + j++; \ + } \ + } while (0) + + + +/* + * Place tuple and split page, original buffer(lbuf) leaves untouched, + * returns shadow page of lbuf filled new data. + */ +static Page +dataSplitPageLeaf(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, + XLogRecData **prdata) +{ + OffsetNumber i, j; + Size totalsize = 0, prevTotalsize; + Pointer ptr, copyPtr, copyEndPtr; + Page page; + Page lpage = PageGetTempPageCopy(BufferGetPage(lbuf)); + Page rpage = BufferGetPage(rbuf); + Size pageSize = PageGetPageSize(lpage); + Size maxItemSize = 0; + ItemPointerData iptr, prevIptr, maxLeftIptr; + int totalCount = 0; + int maxItemIndex = btree->curitem; + int separator = 0; + + /* these must be static so they can be returned to caller */ + static XLogRecData rdata[3]; + static ginxlogSplit data; + static char lpageCopy[BLCKSZ]; + + *prdata = rdata; + data.leftChildBlkno = (GinPageIsLeaf(lpage)) ? + InvalidOffsetNumber : GinGetDownlink(btree->entry); + data.updateBlkno = dataPrepareData(btree, lpage, off); + + /* Copy original data of the page */ + memcpy(lpageCopy, lpage, BLCKSZ); + + /* Reinitialize pages */ + GinInitPage(rpage, GinPageGetOpaque(lpage)->flags, pageSize); + GinInitPage(lpage, GinPageGetOpaque(rpage)->flags, pageSize); + + GinPageGetOpaque(lpage)->u.endoffset = GinDataPageGetData(lpage) - lpage; + GinPageGetOpaque(rpage)->u.endoffset = GinDataPageGetData(rpage) - rpage; + + /* Calculate the whole size we're going to place */ + copyPtr = GinDataPageGetData(lpageCopy); + copyEndPtr = lpageCopy + GinPageGetOpaque(lpageCopy)->u.endoffset; + iptr.ip_blkid.bi_hi = 0; + iptr.ip_blkid.bi_lo = 0; + iptr.ip_posid = 0; + i = FirstOffsetNumber; + while (copyPtr < copyEndPtr) + { + if (i == off) + { + prevIptr = iptr; + iptr = btree->items[maxItemIndex]; + + prevTotalsize = totalsize; + totalsize = ginCheckPlaceToDataPageLeaf(&iptr, &prevIptr, totalsize); + + maxItemIndex++; + totalCount++; + maxItemSize = Max(maxItemSize, totalsize - prevTotalsize); + } + + prevIptr = iptr; + copyPtr = ginDataPageLeafReadItemPointer(copyPtr, &iptr); + + prevTotalsize = totalsize; + totalsize = ginCheckPlaceToDataPageLeaf(&iptr, &prevIptr, totalsize); + + totalCount++; + maxItemSize = Max(maxItemSize, totalsize - prevTotalsize); + i++; + } + + Assert(copyPtr <= copyEndPtr); + if (copyPtr == copyEndPtr) + { + prevIptr = iptr; + iptr = btree->items[maxItemIndex]; + if (GinPageRightMost(lpage)) + { + Size newTotalsize; + + /* + * Found how many new item pointer we're going to add using + * worst case assumptions about odd placement and alignment. + */ + while (maxItemIndex < btree->nitem && + (newTotalsize = ginCheckPlaceToDataPageLeaf(&iptr, &prevIptr, totalsize)) < + 2 * GinDataPageSize - 2 * maxItemSize - 2 * MAXIMUM_ALIGNOF + ) { - GinDataPageAddItem(page, btree->items + btree->curitem, off); - off++; - btree->curitem++; + maxItemIndex++; + totalCount++; + maxItemSize = Max(maxItemSize, newTotalsize - totalsize); + totalsize = newTotalsize; + + prevIptr = iptr; + if (maxItemIndex < btree->nitem) + iptr = btree->items[maxItemIndex]; } - data.nitem = btree->curitem - savedPos; - rdata[cnt].len = sizeofitem * data.nitem; } else { - GinDataPageAddItem(page, btree->items + btree->curitem, off); - btree->curitem++; + prevTotalsize = totalsize; + totalsize = ginCheckPlaceToDataPageLeaf(&iptr, &prevIptr, totalsize); + maxItemIndex++; + + totalCount++; + maxItemSize = Max(maxItemSize, totalsize - prevTotalsize); } } - else - GinDataPageAddItem(page, &(btree->pitem), off); + + /* + * Place item pointers with additional information to the pages using + * previous calculations. XXX: what does this do now that I removed the + * additional information stuff from the patch? + */ + ptr = GinDataPageGetData(lpage); + page = lpage; + j = FirstOffsetNumber; + iptr.ip_blkid.bi_hi = 0; + iptr.ip_blkid.bi_lo = 0; + iptr.ip_posid = 0; + prevIptr = iptr; + copyPtr = GinDataPageGetData(lpageCopy); + i = 0; + while (copyPtr < copyEndPtr) + { + if (i == off) + { + while (btree->curitem < maxItemIndex) + { + iptr = btree->items[btree->curitem]; + + ptr = ginDataPageLeafWriteItemPointer(ptr, &iptr, &prevIptr); + Assert(GinDataPageFreeSpacePre(page, ptr) >= 0); + + btree->curitem++; + prevIptr = iptr; + + CHECK_SWITCH_TO_RPAGE; + i++; + } + } + + copyPtr = ginDataPageLeafReadItemPointer(copyPtr, &iptr); + + ptr = ginDataPageLeafWriteItemPointer(ptr, &iptr, &prevIptr); + Assert(GinDataPageFreeSpacePre(page, ptr) >= 0); + + prevIptr = iptr; + + CHECK_SWITCH_TO_RPAGE; + i++; + } + + /* + * If we didn't add the new items yet, they belong at the end. Insert them + * there. + */ + while (btree->curitem < maxItemIndex) + { + iptr = btree->items[btree->curitem]; + + ptr = ginDataPageLeafWriteItemPointer(ptr, &iptr, &prevIptr); + Assert(GinDataPageFreeSpacePre(page, ptr) >= 0); + + btree->curitem++; + + prevIptr = iptr; + + CHECK_SWITCH_TO_RPAGE; + i++; + } + + Assert(page == rpage); + GinPageGetOpaque(rpage)->u.endoffset = ptr - rpage; + + PostingItemSetBlockNumber(&(btree->pitem), BufferGetBlockNumber(lbuf)); + btree->pitem.key = maxLeftIptr; + btree->rightblkno = BufferGetBlockNumber(rbuf); + + *GinDataPageGetRightBound(rpage) = *GinDataPageGetRightBound(lpage); + *GinDataPageGetRightBound(lpage) = maxLeftIptr; + + /* Fill indexes at the end of pages */ + updateItemIndexes(lpage, btree->ginstate); + updateItemIndexes(rpage, btree->ginstate); + + data.node = btree->index->rd_node; + data.rootBlkno = InvalidBlockNumber; + data.lblkno = BufferGetBlockNumber(lbuf); + data.rblkno = BufferGetBlockNumber(rbuf); + data.separator = separator; + data.nitem = i; + data.isData = TRUE; + data.isLeaf = TRUE; + data.isRootSplit = FALSE; + data.rightbound = *GinDataPageGetRightBound(rpage); + + rdata[0].buffer = InvalidBuffer; + rdata[0].data = (char *) &data; + rdata[0].len = sizeof(ginxlogSplit); + rdata[0].next = &rdata[1]; + + rdata[1].buffer = InvalidBuffer; + rdata[1].data = GinDataPageGetData(lpage); + rdata[1].len = GinDataPageSize - GinLeafDataPageGetFreeSpace(lpage); + rdata[1].next = &rdata[2]; + + rdata[2].buffer = InvalidBuffer; + rdata[2].data = GinDataPageGetData(rpage); + rdata[2].len = GinDataPageSize - GinLeafDataPageGetFreeSpace(rpage); + rdata[2].next = NULL; + + return lpage; } /* @@ -442,19 +895,18 @@ dataPlaceToPage(GinBtree btree, Buffer buf, OffsetNumber off, XLogRecData **prda * left page */ static Page -dataSplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, XLogRecData **prdata) +dataSplitPageInternal(GinBtree btree, Buffer lbuf, Buffer rbuf, + OffsetNumber off, XLogRecData **prdata) { char *ptr; OffsetNumber separator; ItemPointer bound; Page lpage = PageGetTempPageCopy(BufferGetPage(lbuf)); ItemPointerData oldbound = *GinDataPageGetRightBound(lpage); - int sizeofitem = GinSizeOfDataPageItem(lpage); - OffsetNumber maxoff = GinPageGetOpaque(lpage)->maxoff; + OffsetNumber maxoff = GinPageGetOpaque(lpage)->u.maxoff; Page rpage = BufferGetPage(rbuf); Size pageSize = PageGetPageSize(lpage); Size freeSpace; - uint32 nCopied = 1; /* these must be static so they can be returned to caller */ static ginxlogSplit data; @@ -462,7 +914,7 @@ dataSplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, XLogRe static char vector[2 * BLCKSZ]; GinInitPage(rpage, GinPageGetOpaque(lpage)->flags, pageSize); - freeSpace = GinDataPageGetFreeSpace(rpage); + freeSpace = GinNonLeafDataPageGetFreeSpace(rpage); *prdata = rdata; data.leftChildBlkno = (GinPageIsLeaf(lpage)) ? @@ -470,63 +922,36 @@ dataSplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, XLogRe data.updateBlkno = dataPrepareData(btree, lpage, off); memcpy(vector, GinDataPageGetItem(lpage, FirstOffsetNumber), - maxoff * sizeofitem); + maxoff * sizeof(PostingItem)); - if (GinPageIsLeaf(lpage) && GinPageRightMost(lpage) && off > GinPageGetOpaque(lpage)->maxoff) - { - nCopied = 0; - while (btree->curitem < btree->nitem && - maxoff * sizeof(ItemPointerData) < 2 * (freeSpace - sizeof(ItemPointerData))) - { - memcpy(vector + maxoff * sizeof(ItemPointerData), - btree->items + btree->curitem, - sizeof(ItemPointerData)); - maxoff++; - nCopied++; - btree->curitem++; - } - } - else - { - ptr = vector + (off - 1) * sizeofitem; - if (maxoff + 1 - off != 0) - memmove(ptr + sizeofitem, ptr, (maxoff - off + 1) * sizeofitem); - if (GinPageIsLeaf(lpage)) - { - memcpy(ptr, btree->items + btree->curitem, sizeofitem); - btree->curitem++; - } - else - memcpy(ptr, &(btree->pitem), sizeofitem); + ptr = vector + (off - 1) * sizeof(PostingItem); + if (maxoff + 1 - off != 0) + memmove(ptr + sizeof(PostingItem), ptr, (maxoff - off + 1) * sizeof(PostingItem)); + memcpy(ptr, &(btree->pitem), sizeof(PostingItem)); - maxoff++; - } + maxoff++; /* * we suppose that during index creation table scaned from begin to end, * so ItemPointers are monotonically increased.. */ if (btree->isBuild && GinPageRightMost(lpage)) - separator = freeSpace / sizeofitem; + separator = freeSpace / sizeof(PostingItem); else separator = maxoff / 2; GinInitPage(rpage, GinPageGetOpaque(lpage)->flags, pageSize); GinInitPage(lpage, GinPageGetOpaque(rpage)->flags, pageSize); - memcpy(GinDataPageGetItem(lpage, FirstOffsetNumber), vector, separator * sizeofitem); - GinPageGetOpaque(lpage)->maxoff = separator; + memcpy(GinDataPageGetItem(lpage, FirstOffsetNumber), vector, separator * sizeof(PostingItem)); + GinPageGetOpaque(lpage)->u.maxoff = separator; memcpy(GinDataPageGetItem(rpage, FirstOffsetNumber), - vector + separator * sizeofitem, (maxoff - separator) * sizeofitem); - GinPageGetOpaque(rpage)->maxoff = maxoff - separator; + vector + separator * sizeof(PostingItem), (maxoff - separator) * sizeof(PostingItem)); + GinPageGetOpaque(rpage)->u.maxoff = maxoff - separator; PostingItemSetBlockNumber(&(btree->pitem), BufferGetBlockNumber(lbuf)); - if (GinPageIsLeaf(lpage)) - btree->pitem.key = *(ItemPointerData *) GinDataPageGetItem(lpage, - GinPageGetOpaque(lpage)->maxoff); - else - btree->pitem.key = ((PostingItem *) GinDataPageGetItem(lpage, - GinPageGetOpaque(lpage)->maxoff))->key; + btree->pitem.key = ((PostingItem *) GinDataPageGetItem(lpage, + GinPageGetOpaque(lpage)->u.maxoff))->key; btree->rightblkno = BufferGetBlockNumber(rbuf); /* set up right bound for left page */ @@ -544,7 +969,7 @@ dataSplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, XLogRe data.separator = separator; data.nitem = maxoff; data.isData = TRUE; - data.isLeaf = GinPageIsLeaf(lpage) ? TRUE : FALSE; + data.isLeaf = FALSE; data.isRootSplit = FALSE; data.rightbound = oldbound; @@ -555,13 +980,83 @@ dataSplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, XLogRe rdata[1].buffer = InvalidBuffer; rdata[1].data = vector; - rdata[1].len = MAXALIGN(maxoff * sizeofitem); + rdata[1].len = MAXALIGN(maxoff * sizeof(PostingItem)); rdata[1].next = NULL; return lpage; } /* + * Split page of posting tree. Calls relevant function of internal of leaf page + * because they are handled very different. + */ +static Page +dataSplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, + XLogRecData **prdata) +{ + if (GinPageIsLeaf(BufferGetPage(lbuf))) + return dataSplitPageLeaf(btree, lbuf, rbuf, off, prdata); + else + return dataSplitPageInternal(btree, lbuf, rbuf, off, prdata); +} + +/* + * Updates indexes in the end of leaf page which are used for faster search. + * Also updates freespace opaque field of page. Returns last item pointer of + * page. + */ +ItemPointerData +updateItemIndexes(Page page, GinState *ginstate) +{ + Pointer beginptr; + Pointer ptr; + Pointer endptr; + Pointer nextptr; + ItemPointerData iptr; + int j = 0, i; + int totalsize; + + Assert(GinPageIsLeaf(page)); + + /* Iterate over page */ + + ptr = beginptr = GinDataPageGetData(page); + endptr = page + GinPageGetOpaque(page)->u.endoffset; + iptr.ip_blkid.bi_lo = 0; + iptr.ip_blkid.bi_hi = 0; + iptr.ip_posid = 0; + + totalsize = endptr - beginptr; + nextptr = beginptr + (int) ((double) totalsize / (double) (GinDataLeafIndexCount + 1)); + j = 0; + i = FirstOffsetNumber; + while (ptr < endptr && j < GinDataLeafIndexCount) + { + /* Place next page index entry if it's time to */ + if (ptr >= nextptr) + { + GinPageGetIndexes(page)[j].iptr = iptr; + GinPageGetIndexes(page)[j].offsetNumber = i; + GinPageGetIndexes(page)[j].pageOffset = ptr - GinDataPageGetData(page); + j++; + nextptr = beginptr + (int) ((double) (j + 1) * (double) totalsize / (double) (GinDataLeafIndexCount + 1)); + } + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + i++; + } + /* Fill rest of page indexes with InvalidOffsetNumber if any */ + for (; j < GinDataLeafIndexCount; j++) + { + GinDataLeafItemIndex *idx = &GinPageGetIndexes(page)[j]; + MemSet(&idx->iptr, 0, sizeof(ItemPointerData)); + idx->offsetNumber = InvalidOffsetNumber; + idx->pageOffset = 0; + } + + return iptr; +} + +/* * Fills new root by right bound values from child. * Also called from ginxlog, should not use btree */ @@ -576,11 +1071,11 @@ ginDataFillRoot(GinBtree btree, Buffer root, Buffer lbuf, Buffer rbuf) li.key = *GinDataPageGetRightBound(lpage); PostingItemSetBlockNumber(&li, BufferGetBlockNumber(lbuf)); - GinDataPageAddItem(page, &li, InvalidOffsetNumber); + GinPageAddPostingItem(page, &li, InvalidOffsetNumber); ri.key = *GinDataPageGetRightBound(rpage); PostingItemSetBlockNumber(&ri, BufferGetBlockNumber(rbuf)); - GinDataPageAddItem(page, &ri, InvalidOffsetNumber); + GinPageAddPostingItem(page, &ri, InvalidOffsetNumber); } void diff --git a/src/backend/access/gin/ginentrypage.c b/src/backend/access/gin/ginentrypage.c index 7733028..f069ea6 100644 --- a/src/backend/access/gin/ginentrypage.c +++ b/src/backend/access/gin/ginentrypage.c @@ -18,151 +18,24 @@ #include "utils/rel.h" /* - * Form a tuple for entry tree. - * - * If the tuple would be too big to be stored, function throws a suitable - * error if errorTooBig is TRUE, or returns NULL if errorTooBig is FALSE. - * - * See src/backend/access/gin/README for a description of the index tuple - * format that is being built here. We build on the assumption that we - * are making a leaf-level key entry containing a posting list of nipd items. - * If the caller is actually trying to make a posting-tree entry, non-leaf - * entry, or pending-list entry, it should pass nipd = 0 and then overwrite - * the t_tid fields as necessary. In any case, ipd can be NULL to skip - * copying any itempointers into the posting list; the caller is responsible - * for filling the posting list afterwards, if ipd = NULL and nipd > 0. + * Read item pointers from leaf data page. Information is stored in the same + * manner as in leaf data pages. */ -IndexTuple -GinFormTuple(GinState *ginstate, - OffsetNumber attnum, Datum key, GinNullCategory category, - ItemPointerData *ipd, uint32 nipd, - bool errorTooBig) +void +ginReadTuple(GinState *ginstate, OffsetNumber attnum, + IndexTuple itup, ItemPointerData *ipd) { - Datum datums[2]; - bool isnull[2]; - IndexTuple itup; - uint32 newsize; - - /* Build the basic tuple: optional column number, plus key datum */ - if (ginstate->oneCol) - { - datums[0] = key; - isnull[0] = (category != GIN_CAT_NORM_KEY); - } - else - { - datums[0] = UInt16GetDatum(attnum); - isnull[0] = false; - datums[1] = key; - isnull[1] = (category != GIN_CAT_NORM_KEY); - } - - itup = index_form_tuple(ginstate->tupdesc[attnum - 1], datums, isnull); - - /* - * Determine and store offset to the posting list, making sure there is - * room for the category byte if needed. - * - * Note: because index_form_tuple MAXALIGNs the tuple size, there may well - * be some wasted pad space. Is it worth recomputing the data length to - * prevent that? That would also allow us to Assert that the real data - * doesn't overlap the GinNullCategory byte, which this code currently - * takes on faith. - */ - newsize = IndexTupleSize(itup); - - if (IndexTupleHasNulls(itup)) - { - uint32 minsize; - - Assert(category != GIN_CAT_NORM_KEY); - minsize = GinCategoryOffset(itup, ginstate) + sizeof(GinNullCategory); - newsize = Max(newsize, minsize); - } - - newsize = SHORTALIGN(newsize); - - GinSetPostingOffset(itup, newsize); - - GinSetNPosting(itup, nipd); - - /* - * Add space needed for posting list, if any. Then check that the tuple - * won't be too big to store. - */ - newsize += sizeof(ItemPointerData) * nipd; - newsize = MAXALIGN(newsize); - if (newsize > Min(INDEX_SIZE_MASK, GinMaxItemSize)) - { - if (errorTooBig) - ereport(ERROR, - (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), - errmsg("index row size %lu exceeds maximum %lu for index \"%s\"", - (unsigned long) newsize, - (unsigned long) Min(INDEX_SIZE_MASK, - GinMaxItemSize), - RelationGetRelationName(ginstate->index)))); - pfree(itup); - return NULL; - } + Pointer ptr; + int nipd = GinGetNPosting(itup), i; + ItemPointerData ip = {{0,0},0}; - /* - * Resize tuple if needed - */ - if (newsize != IndexTupleSize(itup)) - { - itup = repalloc(itup, newsize); - /* - * PostgreSQL 9.3 and earlier did not clear this new space, so we - * might find uninitialized padding when reading tuples from disk. - */ - memset((char *) itup + IndexTupleSize(itup), - 0, newsize - IndexTupleSize(itup)); + ptr = GinGetPosting(itup); - /* set new size in tuple header */ - itup->t_info &= ~INDEX_SIZE_MASK; - itup->t_info |= newsize; - } - - /* - * Insert category byte, if needed - */ - if (category != GIN_CAT_NORM_KEY) + for (i = 0; i < nipd; i++) { - Assert(IndexTupleHasNulls(itup)); - GinSetNullCategory(itup, ginstate, category); + ptr = ginDataPageLeafReadItemPointer(ptr, &ip); + ipd[i] = ip; } - - /* - * Copy in the posting list, if provided - */ - if (ipd) - memcpy(GinGetPosting(itup), ipd, sizeof(ItemPointerData) * nipd); - - return itup; -} - -/* - * Sometimes we reduce the number of posting list items in a tuple after - * having built it with GinFormTuple. This function adjusts the size - * fields to match. - */ -void -GinShortenTuple(IndexTuple itup, uint32 nipd) -{ - uint32 newsize; - - Assert(nipd <= GinGetNPosting(itup)); - - newsize = GinGetPostingOffset(itup) + sizeof(ItemPointerData) * nipd; - newsize = MAXALIGN(newsize); - - Assert(newsize <= (itup->t_info & INDEX_SIZE_MASK)); - - itup->t_info &= ~INDEX_SIZE_MASK; - itup->t_info |= newsize; - - GinSetNPosting(itup, nipd); } /* diff --git a/src/backend/access/gin/ginfast.c b/src/backend/access/gin/ginfast.c index 4f2c118..49f799d 100644 --- a/src/backend/access/gin/ginfast.c +++ b/src/backend/access/gin/ginfast.c @@ -23,6 +23,7 @@ #include "miscadmin.h" #include "utils/memutils.h" #include "utils/rel.h" +#include "access/htup_details.h" #define GIN_PAGE_FREESIZE \ @@ -94,11 +95,11 @@ writeListPage(Relation index, Buffer buffer, if (rightlink == InvalidBlockNumber) { GinPageSetFullRow(page); - GinPageGetOpaque(page)->maxoff = 1; + GinPageGetOpaque(page)->u.maxoff = 1; } else { - GinPageGetOpaque(page)->maxoff = 0; + GinPageGetOpaque(page)->u.maxoff = 0; } MarkBufferDirty(buffer); @@ -368,8 +369,8 @@ ginHeapTupleFastInsert(GinState *ginstate, GinTupleCollector *collector) /* * Increase counter of heap tuples */ - Assert(GinPageGetOpaque(page)->maxoff <= metadata->nPendingHeapTuples); - GinPageGetOpaque(page)->maxoff++; + Assert(GinPageGetOpaque(page)->u.maxoff <= metadata->nPendingHeapTuples); + GinPageGetOpaque(page)->u.maxoff++; metadata->nPendingHeapTuples++; for (i = 0; i < collector->ntuples; i++) @@ -437,6 +438,89 @@ ginHeapTupleFastInsert(GinState *ginstate, GinTupleCollector *collector) ginInsertCleanup(ginstate, false, NULL); } +static IndexTuple +GinFastFormTuple(GinState *ginstate, + OffsetNumber attnum, Datum key, GinNullCategory category) +{ + Datum datums[2]; + bool isnull[2]; + IndexTuple itup; + uint32 newsize; + + /* Build the basic tuple: optional column number, plus key datum */ + + if (ginstate->oneCol) + { + datums[0] = key; + isnull[0] = (category != GIN_CAT_NORM_KEY); + } + else + { + datums[0] = UInt16GetDatum(attnum); + isnull[0] = false; + datums[1] = key; + isnull[1] = (category != GIN_CAT_NORM_KEY); + } + + itup = index_form_tuple(ginstate->tupdesc[attnum - 1], datums, isnull); + + /* + * Place category to the last byte of index tuple extending it's size if + * needed + */ + newsize = IndexTupleSize(itup); + + if (category != GIN_CAT_NORM_KEY) + { + uint32 minsize; + + Assert(IndexTupleHasNulls(itup)); + minsize = IndexInfoFindDataOffset(itup->t_info) + + heap_compute_data_size(ginstate->tupdesc[attnum - 1], datums, isnull) + + sizeof(GinNullCategory); + newsize = Max(newsize, minsize); + } + + newsize = MAXALIGN(newsize); + + if (newsize > Min(INDEX_SIZE_MASK, GinMaxItemSize)) + { + ereport(ERROR, + (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), + errmsg("index row size %lu exceeds maximum %lu for index \"%s\"", + (unsigned long) newsize, + (unsigned long) Min(INDEX_SIZE_MASK, + GinMaxItemSize), + RelationGetRelationName(ginstate->index)))); + pfree(itup); + return NULL; + } + + /* + * Resize tuple if needed + */ + if (newsize != IndexTupleSize(itup)) + { + itup = repalloc(itup, newsize); + + /* set new size in tuple header */ + itup->t_info &= ~INDEX_SIZE_MASK; + itup->t_info |= newsize; + } + + /* + * Insert category byte, if needed + */ + if (category != GIN_CAT_NORM_KEY) + { + Assert(IndexTupleHasNulls(itup)); + GinSetNullCategory(itup, ginstate, category); + } + + return itup; +} + + /* * Create temporary index tuples for a single indexable item (one index column * for the heap tuple specified by ht_ctid), and append them to the array @@ -486,8 +570,7 @@ ginHeapTupleFastCollect(GinState *ginstate, { IndexTuple itup; - itup = GinFormTuple(ginstate, attnum, entries[i], categories[i], - NULL, 0, true); + itup = GinFastFormTuple(ginstate, attnum, entries[i], categories[i]); itup->t_tid = *ht_ctid; collector->tuples[collector->ntuples++] = itup; collector->sumsize += IndexTupleSize(itup); @@ -550,7 +633,7 @@ shiftList(Relation index, Buffer metabuffer, BlockNumber newHead, return true; } - nDeletedHeapTuples += GinPageGetOpaque(page)->maxoff; + nDeletedHeapTuples += GinPageGetOpaque(page)->u.maxoff; blknoToDelete = GinPageGetOpaque(page)->rightlink; } diff --git a/src/backend/access/gin/ginget.c b/src/backend/access/gin/ginget.c index cb17d38..7422f52 100644 --- a/src/backend/access/gin/ginget.c +++ b/src/backend/access/gin/ginget.c @@ -71,20 +71,25 @@ callConsistentFn(GinState *ginstate, GinScanKey key) static bool findItemInPostingPage(Page page, ItemPointer item, OffsetNumber *off) { - OffsetNumber maxoff = GinPageGetOpaque(page)->maxoff; int res; + Pointer ptr; + Pointer endPtr; + ItemPointerData iptr = {{0, 0}, 0}; if (GinPageGetOpaque(page)->flags & GIN_DELETED) /* page was deleted by concurrent vacuum */ return false; + ptr = GinDataPageGetData(page); + endPtr = page + GinPageGetOpaque(page)->u.endoffset; /* * scan page to find equal or first greater value */ - for (*off = FirstOffsetNumber; *off <= maxoff; (*off)++) + while (ptr < endPtr) { - res = ginCompareItemPointers(item, (ItemPointer) GinDataPageGetItem(page, *off)); + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + res = ginCompareItemPointers(item, &iptr); if (res <= 0) return true; } @@ -148,15 +153,27 @@ scanPostingTree(Relation index, GinScanEntry scanEntry, */ for (;;) { - page = BufferGetPage(buffer); + Pointer ptr; + Pointer endPtr; + page = BufferGetPage(buffer); + ptr = GinDataPageGetData(page); + endPtr = page + GinPageGetOpaque(page)->u.endoffset; if ((GinPageGetOpaque(page)->flags & GIN_DELETED) == 0 && - GinPageGetOpaque(page)->maxoff >= FirstOffsetNumber) + ptr < endPtr) { - tbm_add_tuples(scanEntry->matchBitmap, - (ItemPointer) GinDataPageGetItem(page, FirstOffsetNumber), - GinPageGetOpaque(page)->maxoff, false); - scanEntry->predictNumberResult += GinPageGetOpaque(page)->maxoff; + ItemPointerData iptr = {{0, 0}, 0}; + int i; + + i = 0; + while (ptr < endPtr) + { + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + tbm_add_tuples(scanEntry->matchBitmap, &iptr, 1, false); + i++; + } + + scanEntry->predictNumberResult += i; } if (GinPageRightMost(page)) @@ -344,8 +361,12 @@ collectMatchBitmap(GinBtreeData *btree, GinBtreeStack *stack, } else { + ItemPointerData *ipd = (ItemPointerData *)palloc( + sizeof(ItemPointerData) * GinGetNPosting(itup)); + ginReadTuple(btree->ginstate, scanEntry->attnum, itup, ipd); + tbm_add_tuples(scanEntry->matchBitmap, - GinGetPosting(itup), GinGetNPosting(itup), false); + ipd, GinGetNPosting(itup), false); scanEntry->predictNumberResult += GinGetNPosting(itup); } @@ -438,6 +459,9 @@ restartScanEntry: BlockNumber rootPostingTree = GinGetPostingTree(itup); GinPostingTreeScan *gdi; Page page; + Pointer ptr; + Pointer endPtr; + ItemPointerData iptr = {{0,0},0}; /* * We should unlock entry page before touching posting tree to @@ -460,15 +484,22 @@ restartScanEntry: IncrBufferRefCount(entry->buffer); page = BufferGetPage(entry->buffer); - entry->predictNumberResult = gdi->stack->predictNumber * GinPageGetOpaque(page)->maxoff; /* * Keep page content in memory to prevent durable page locking */ - entry->list = (ItemPointerData *) palloc(BLCKSZ); - entry->nlist = GinPageGetOpaque(page)->maxoff; - memcpy(entry->list, GinDataPageGetItem(page, FirstOffsetNumber), - GinPageGetOpaque(page)->maxoff * sizeof(ItemPointerData)); + entry->list = (ItemPointerData *) palloc(BLCKSZ * sizeof(ItemPointerData)); + entry->nlist = 0; + + ptr = GinDataPageGetData(page); + endPtr = page + GinPageGetOpaque(page)->u.endoffset; + while (ptr < endPtr) + { + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + entry->list[entry->nlist++] = iptr; + } + + entry->predictNumberResult = gdi->stack->predictNumber * entry->nlist; LockBuffer(entry->buffer, GIN_UNLOCK); freeGinBtreeStack(gdi->stack); @@ -478,8 +509,11 @@ restartScanEntry: else if (GinGetNPosting(itup) > 0) { entry->nlist = GinGetNPosting(itup); + entry->predictNumberResult = entry->nlist; entry->list = (ItemPointerData *) palloc(sizeof(ItemPointerData) * entry->nlist); - memcpy(entry->list, GinGetPosting(itup), sizeof(ItemPointerData) * entry->nlist); + + ginReadTuple(ginstate, entry->attnum, itup, entry->list); + entry->isFinished = FALSE; } } @@ -583,12 +617,21 @@ entryGetNextItem(GinState *ginstate, GinScanEntry entry) if (!ItemPointerIsValid(&entry->curItem) || findItemInPostingPage(page, &entry->curItem, &entry->offset)) { + Pointer ptr; + Pointer endPtr; + ItemPointerData iptr = {{0,0},0}; + /* * Found position equal to or greater than stored */ - entry->nlist = GinPageGetOpaque(page)->maxoff; - memcpy(entry->list, GinDataPageGetItem(page, FirstOffsetNumber), - GinPageGetOpaque(page)->maxoff * sizeof(ItemPointerData)); + ptr = GinDataPageGetData(page); + endPtr = page + GinPageGetOpaque(page)->u.endoffset; + entry->nlist = 0; + while (ptr < endPtr) + { + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + entry->list[entry->nlist++] = iptr; + } LockBuffer(entry->buffer, GIN_UNLOCK); diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index beaa653..6177290 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -42,14 +42,16 @@ typedef struct * items[] must be in sorted order with no duplicates. */ static BlockNumber -createPostingTree(Relation index, ItemPointerData *items, uint32 nitems) +createPostingTree(GinState *ginstate, ItemPointerData *items, uint32 nitems) { BlockNumber blkno; - Buffer buffer = GinNewBuffer(index); + Buffer buffer = GinNewBuffer(ginstate->index); Page page; + int i; + Pointer ptr; + ItemPointerData prev_iptr = {{0,0},0}; /* Assert that the items[] array will fit on one page */ - Assert(nitems <= GinMaxLeafDataItems); START_CRIT_SECTION(); @@ -57,18 +59,26 @@ createPostingTree(Relation index, ItemPointerData *items, uint32 nitems) page = BufferGetPage(buffer); blkno = BufferGetBlockNumber(buffer); - memcpy(GinDataPageGetData(page), items, sizeof(ItemPointerData) * nitems); - GinPageGetOpaque(page)->maxoff = nitems; + ptr = GinDataPageGetData(page); + for (i = 0; i < nitems; i++) + { + if (i > 0) + prev_iptr = items[i - 1]; + ptr = ginDataPageLeafWriteItemPointer(ptr, &items[i], &prev_iptr); + } + GinPageGetOpaque(page)->u.endoffset = ptr - page; + Assert(GinDataPageFreeSpacePre(page, ptr) >= 0); + updateItemIndexes(page, ginstate); MarkBufferDirty(buffer); - if (RelationNeedsWAL(index)) + if (RelationNeedsWAL(ginstate->index)) { XLogRecPtr recptr; XLogRecData rdata[2]; ginxlogCreatePostingTree data; - data.node = index->rd_node; + data.node = ginstate->index->rd_node; data.blkno = blkno; data.nitem = nitems; @@ -78,8 +88,8 @@ createPostingTree(Relation index, ItemPointerData *items, uint32 nitems) rdata[0].next = &rdata[1]; rdata[1].buffer = InvalidBuffer; - rdata[1].data = (char *) items; - rdata[1].len = sizeof(ItemPointerData) * nitems; + rdata[1].data = GinDataPageGetData(page); + rdata[1].len = ptr - GinDataPageGetData(page); rdata[1].next = NULL; recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_PTREE, rdata); @@ -93,6 +103,137 @@ createPostingTree(Relation index, ItemPointerData *items, uint32 nitems) return blkno; } +/* + * Form a tuple for entry tree. + * + * If the tuple would be too big to be stored, function throws a suitable + * error if errorTooBig is TRUE, or returns NULL if errorTooBig is FALSE. + * + * See src/backend/access/gin/README for a description of the index tuple + * format that is being built here. We build on the assumption that we + * are making a leaf-level key entry containing a posting list of nipd items. + * If the caller is actually trying to make a posting-tree entry, non-leaf + * entry, or pending-list entry, it should pass nipd = 0 and then overwrite + * the t_tid fields as necessary. In any case, ipd can be NULL to skip + * copying any itempointers into the posting list; the caller is responsible + * for filling the posting list afterwards, if ipd = NULL and nipd > 0. + */ +static IndexTuple +GinFormTuple(GinState *ginstate, + OffsetNumber attnum, Datum key, GinNullCategory category, + ItemPointerData *ipd, + uint32 nipd, + bool errorTooBig) +{ + Datum datums[3]; + bool isnull[3]; + IndexTuple itup; + uint32 newsize; + int i; + ItemPointerData nullItemPointer = {{0,0},0}; + + /* Build the basic tuple: optional column number, plus key datum */ + if (ginstate->oneCol) + { + datums[0] = key; + isnull[0] = (category != GIN_CAT_NORM_KEY); + isnull[1] = true; + } + else + { + datums[0] = UInt16GetDatum(attnum); + isnull[0] = false; + datums[1] = key; + isnull[1] = (category != GIN_CAT_NORM_KEY); + isnull[2] = true; + } + + itup = index_form_tuple(ginstate->tupdesc[attnum - 1], datums, isnull); + + /* + * Determine and store offset to the posting list, making sure there is + * room for the category byte if needed. + * + * Note: because index_form_tuple MAXALIGNs the tuple size, there may well + * be some wasted pad space. Is it worth recomputing the data length to + * prevent that? That would also allow us to Assert that the real data + * doesn't overlap the GinNullCategory byte, which this code currently + * takes on faith. + */ + newsize = IndexTupleSize(itup); + + GinSetPostingOffset(itup, newsize); + + GinSetNPosting(itup, nipd); + + /* + * Add space needed for posting list, if any. Then check that the tuple + * won't be too big to store. + */ + + if (nipd > 0) + { + newsize = ginCheckPlaceToDataPageLeaf(&ipd[0], &nullItemPointer, newsize); + for (i = 1; i < nipd; i++) + { + newsize = ginCheckPlaceToDataPageLeaf(&ipd[i], &ipd[i - 1], newsize); + } + } + + if (category != GIN_CAT_NORM_KEY) + { + Assert(IndexTupleHasNulls(itup)); + newsize = newsize + sizeof(GinNullCategory); + } + newsize = MAXALIGN(newsize); + + if (newsize > Min(INDEX_SIZE_MASK, GinMaxItemSize)) + { + if (errorTooBig) + ereport(ERROR, + (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), + errmsg("index row size %lu exceeds maximum %lu for index \"%s\"", + (unsigned long) newsize, + (unsigned long) Min(INDEX_SIZE_MASK, + GinMaxItemSize), + RelationGetRelationName(ginstate->index)))); + pfree(itup); + return NULL; + } + + /* + * Resize tuple if needed + */ + if (newsize != IndexTupleSize(itup)) + { + itup = repalloc(itup, newsize); + + /* set new size in tuple header */ + itup->t_info &= ~INDEX_SIZE_MASK; + itup->t_info |= newsize; + } + + /* + * Copy in the posting list, if provided + */ + if (nipd > 0) + { + char *ptr = GinGetPosting(itup); + ptr = ginDataPageLeafWriteItemPointer(ptr, &ipd[0], &nullItemPointer); + for (i = 1; i < nipd; i++) + ptr = ginDataPageLeafWriteItemPointer(ptr, &ipd[i], &ipd[i-1]); + } + + /* + * Insert category byte, if needed + */ + if (category != GIN_CAT_NORM_KEY) + { + Assert(IndexTupleHasNulls(itup)); + GinSetNullCategory(itup, ginstate, category); + } + return itup; +} /* * Adds array of item pointers to tuple's posting list, or @@ -111,31 +252,30 @@ addItemPointersToLeafTuple(GinState *ginstate, Datum key; GinNullCategory category; IndexTuple res; + ItemPointerData *newItems, *oldItems; + int oldNPosting, newNPosting; Assert(!GinIsPostingTree(old)); attnum = gintuple_get_attrnum(ginstate, old); key = gintuple_get_key(ginstate, old, &category); + oldNPosting = GinGetNPosting(old); + oldItems = (ItemPointerData *) palloc(sizeof(ItemPointerData) * oldNPosting); + + newNPosting = oldNPosting + nitem; + newItems = (ItemPointerData *) palloc(sizeof(ItemPointerData) * newNPosting); + + ginReadTuple(ginstate, attnum, old, oldItems); + + newNPosting = ginMergeItemPointers(newItems, + items, nitem, + oldItems, oldNPosting); + /* try to build tuple with room for all the items */ res = GinFormTuple(ginstate, attnum, key, category, - NULL, nitem + GinGetNPosting(old), - false); - - if (res) - { - /* good, small enough */ - uint32 newnitem; - - /* fill in the posting list with union of old and new TIDs */ - newnitem = ginMergeItemPointers(GinGetPosting(res), - GinGetPosting(old), - GinGetNPosting(old), - items, nitem); - /* merge might have eliminated some duplicate items */ - GinShortenTuple(res, newnitem); - } - else + newItems, newNPosting, false); + if (!res) { /* posting list would be too big, convert to posting tree */ BlockNumber postingRoot; @@ -146,9 +286,9 @@ addItemPointersToLeafTuple(GinState *ginstate, * surely small enough to fit on one posting-tree page, and should * already be in order with no duplicates. */ - postingRoot = createPostingTree(ginstate->index, - GinGetPosting(old), - GinGetNPosting(old)); + postingRoot = createPostingTree(ginstate, + oldItems, + oldNPosting); /* During index build, count the newly-added data page */ if (buildStats) @@ -194,6 +334,19 @@ buildFreshLeafTuple(GinState *ginstate, { /* posting list would be too big, build posting tree */ BlockNumber postingRoot; + ItemPointerData prevIptr = {{0,0},0}; + Size size = 0; + int itemsCount = 0; + + do + { + size = ginCheckPlaceToDataPageLeaf(&items[itemsCount], &prevIptr, + size); + prevIptr = items[itemsCount]; + itemsCount++; + } + while (itemsCount < nitem && size < GinDataPageSize); + itemsCount--; /* * Build posting-tree-only result tuple. We do this first so as to @@ -205,16 +358,16 @@ buildFreshLeafTuple(GinState *ginstate, * Initialize posting tree with as many TIDs as will fit on the first * page. */ - postingRoot = createPostingTree(ginstate->index, + postingRoot = createPostingTree(ginstate, items, - Min(nitem, GinMaxLeafDataItems)); + itemsCount); /* During index build, count the newly-added data page */ if (buildStats) buildStats->nDataPages++; /* Add any remaining TIDs to the posting tree */ - if (nitem > GinMaxLeafDataItems) + if (nitem > itemsCount) { GinPostingTreeScan *gdi; @@ -222,8 +375,8 @@ buildFreshLeafTuple(GinState *ginstate, gdi->btree.isBuild = (buildStats != NULL); ginInsertItemPointers(gdi, - items + GinMaxLeafDataItems, - nitem - GinMaxLeafDataItems, + items + itemsCount, + nitem - itemsCount, buildStats); pfree(gdi); diff --git a/src/backend/access/gin/ginvacuum.c b/src/backend/access/gin/ginvacuum.c index b84d3a3..5e77ac5 100644 --- a/src/backend/access/gin/ginvacuum.c +++ b/src/backend/access/gin/ginvacuum.c @@ -38,47 +38,177 @@ typedef struct * if it's needed. In case of *cleaned!=NULL caller is responsible to * have allocated enough space. *cleaned and items may point to the same * memory address. + * + * The caller can specify the size of 'src' */ static uint32 -ginVacuumPostingList(GinVacuumState *gvs, ItemPointerData *items, uint32 nitem, ItemPointerData **cleaned) +ginVacuumPostingList(GinVacuumState *gvs, + Pointer src, uint32 nitem, Pointer srcEnd, + Pointer *cleaned, Size size, Size *newSize) { uint32 i, j = 0; + ItemPointerData iptr = {{0,0},0}, prevIptr; + Pointer dst = NULL, prev, ptr = src; /* * just scan over ItemPointer array */ - for (i = 0; i < nitem; i++) + prevIptr = iptr; + for (i = 0; srcEnd ? (ptr < srcEnd) : (i < nitem); i++) { - if (gvs->callback(items + i, gvs->callback_state)) + prev = ptr; + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + if (gvs->callback(&iptr, gvs->callback_state)) { gvs->result->tuples_removed += 1; - if (!*cleaned) + if (!dst) { - *cleaned = (ItemPointerData *) palloc(sizeof(ItemPointerData) * nitem); + dst = (Pointer) palloc(size); + *cleaned = dst; if (i != 0) - memcpy(*cleaned, items, sizeof(ItemPointerData) * i); + { + memcpy(dst, src, prev - src); + dst += prev - src; + } } } else { gvs->result->num_index_tuples += 1; if (i != j) - (*cleaned)[j] = items[i]; + dst = ginDataPageLeafWriteItemPointer(dst, &iptr, &prevIptr); j++; + prevIptr = iptr; } } + if (i != j) + *newSize = dst - *cleaned; return j; } /* + * Form a tuple for entry tree based on already encoded array of item pointers + * with additional information. + */ +static IndexTuple +GinFormTuple(GinState *ginstate, + OffsetNumber attnum, Datum key, GinNullCategory category, + Pointer data, + Size dataSize, + uint32 nipd, + bool errorTooBig) +{ + Datum datums[3]; + bool isnull[3]; + IndexTuple itup; + uint32 newsize; + + /* Build the basic tuple: optional column number, plus key datum */ + if (ginstate->oneCol) + { + datums[0] = key; + isnull[0] = (category != GIN_CAT_NORM_KEY); + isnull[1] = true; + } + else + { + datums[0] = UInt16GetDatum(attnum); + isnull[0] = false; + datums[1] = key; + isnull[1] = (category != GIN_CAT_NORM_KEY); + isnull[2] = true; + } + + itup = index_form_tuple(ginstate->tupdesc[attnum - 1], datums, isnull); + + /* + * Determine and store offset to the posting list, making sure there is + * room for the category byte if needed. + * + * Note: because index_form_tuple MAXALIGNs the tuple size, there may well + * be some wasted pad space. Is it worth recomputing the data length to + * prevent that? That would also allow us to Assert that the real data + * doesn't overlap the GinNullCategory byte, which this code currently + * takes on faith. + */ + newsize = IndexTupleSize(itup); + + GinSetPostingOffset(itup, newsize); + + GinSetNPosting(itup, nipd); + + /* + * Add space needed for posting list, if any. Then check that the tuple + * won't be too big to store. + */ + + if (nipd > 0) + { + newsize += dataSize; + } + + if (category != GIN_CAT_NORM_KEY) + { + Assert(IndexTupleHasNulls(itup)); + newsize = newsize + sizeof(GinNullCategory); + } + newsize = MAXALIGN(newsize); + + if (newsize > Min(INDEX_SIZE_MASK, GinMaxItemSize)) + { + if (errorTooBig) + ereport(ERROR, + (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), + errmsg("index row size %lu exceeds maximum %lu for index \"%s\"", + (unsigned long) newsize, + (unsigned long) Min(INDEX_SIZE_MASK, + GinMaxItemSize), + RelationGetRelationName(ginstate->index)))); + pfree(itup); + return NULL; + } + + /* + * Resize tuple if needed + */ + if (newsize != IndexTupleSize(itup)) + { + itup = repalloc(itup, newsize); + + /* set new size in tuple header */ + itup->t_info &= ~INDEX_SIZE_MASK; + itup->t_info |= newsize; + } + + /* + * Copy in the posting list, if provided + */ + if (nipd > 0) + { + char *ptr = GinGetPosting(itup); + memcpy(ptr, data, dataSize); + } + + /* + * Insert category byte, if needed + */ + if (category != GIN_CAT_NORM_KEY) + { + Assert(IndexTupleHasNulls(itup)); + GinSetNullCategory(itup, ginstate, category); + } + return itup; +} + +/* * fills WAL record for vacuum leaf page */ static void -xlogVacuumPage(Relation index, Buffer buffer) +xlogVacuumPage(Relation index, Buffer buffer, OffsetNumber maxoff) { Page page = BufferGetPage(buffer); XLogRecPtr recptr; @@ -95,13 +225,14 @@ xlogVacuumPage(Relation index, Buffer buffer) data.node = index->rd_node; data.blkno = BufferGetBlockNumber(buffer); - + data.nitem = maxoff; if (GinPageIsData(page)) { backup = GinDataPageGetData(page); - data.nitem = GinPageGetOpaque(page)->maxoff; - if (data.nitem) - len = MAXALIGN(sizeof(ItemPointerData) * data.nitem); + if (GinPageIsLeaf(page)) + len = page + GinPageGetOpaque(page)->u.endoffset - GinDataPageGetData(page); + else if (maxoff) + len = MAXALIGN(GinDataPageSize - GinNonLeafDataPageGetFreeSpace(page)); } else { @@ -176,30 +307,37 @@ ginVacuumPostingTreeLeaves(GinVacuumState *gvs, BlockNumber blkno, bool isRoot, if (GinPageIsLeaf(page)) { - OffsetNumber newMaxOff, - oldMaxOff = GinPageGetOpaque(page)->maxoff; - ItemPointerData *cleaned = NULL; + Pointer cleaned = NULL; + Size newSize; + Size oldSize; + Pointer endPtr = page + GinPageGetOpaque(page)->u.endoffset; + OffsetNumber newMaxOff; + oldSize = GinDataPageSize - GinLeafDataPageGetFreeSpace(page); newMaxOff = ginVacuumPostingList(gvs, - (ItemPointer) GinDataPageGetData(page), oldMaxOff, &cleaned); + GinDataPageGetData(page), 0, endPtr, + &cleaned, + oldSize, &newSize); /* saves changes about deleted tuple ... */ - if (oldMaxOff != newMaxOff) + if (cleaned) { START_CRIT_SECTION(); if (newMaxOff > 0) - memcpy(GinDataPageGetData(page), cleaned, sizeof(ItemPointerData) * newMaxOff); + memcpy(GinDataPageGetData(page), cleaned, newSize); + pfree(cleaned); - GinPageGetOpaque(page)->maxoff = newMaxOff; + GinPageGetOpaque(page)->u.endoffset = GinDataPageGetData(page) + newSize - page; + updateItemIndexes(page, &gvs->ginstate); MarkBufferDirty(buffer); - xlogVacuumPage(gvs->index, buffer); + xlogVacuumPage(gvs->index, buffer, newMaxOff); END_CRIT_SECTION(); /* if root is a leaf page, we don't desire further processing */ - if (!isRoot && GinPageGetOpaque(page)->maxoff < FirstOffsetNumber) + if (!isRoot && newMaxOff < FirstOffsetNumber) hasVoidPage = TRUE; } } @@ -208,7 +346,7 @@ ginVacuumPostingTreeLeaves(GinVacuumState *gvs, BlockNumber blkno, bool isRoot, OffsetNumber i; bool isChildHasVoid = FALSE; - for (i = FirstOffsetNumber; i <= GinPageGetOpaque(page)->maxoff; i++) + for (i = FirstOffsetNumber; i <= GinPageGetOpaque(page)->u.maxoff; i++) { PostingItem *pitem = (PostingItem *) GinDataPageGetItem(page, i); @@ -391,6 +529,7 @@ ginScanToDelete(GinVacuumState *gvs, BlockNumber blkno, bool isRoot, DataPageDel Buffer buffer; Page page; bool meDelete = FALSE; + bool isempty; if (isRoot) { @@ -420,7 +559,7 @@ ginScanToDelete(GinVacuumState *gvs, BlockNumber blkno, bool isRoot, DataPageDel OffsetNumber i; me->blkno = blkno; - for (i = FirstOffsetNumber; i <= GinPageGetOpaque(page)->maxoff; i++) + for (i = FirstOffsetNumber; i <= GinPageGetOpaque(page)->u.maxoff; i++) { PostingItem *pitem = (PostingItem *) GinDataPageGetItem(page, i); @@ -429,17 +568,19 @@ ginScanToDelete(GinVacuumState *gvs, BlockNumber blkno, bool isRoot, DataPageDel } } - if (GinPageGetOpaque(page)->maxoff < FirstOffsetNumber) + if (GinPageIsLeaf(page)) + isempty = page + GinPageGetOpaque(page)->u.endoffset == GinDataPageGetData(page); + else + isempty = GinPageGetOpaque(page)->u.maxoff < FirstOffsetNumber; + + if (isempty) { if (!(me->leftBlkno == InvalidBlockNumber && GinPageRightMost(page))) { /* we never delete right most branch */ Assert(!isRoot); - if (GinPageGetOpaque(page)->maxoff < FirstOffsetNumber) - { - ginDeletePage(gvs, blkno, me->leftBlkno, me->parent->blkno, myoff, me->parent->isRoot); - meDelete = TRUE; - } + ginDeletePage(gvs, blkno, me->leftBlkno, me->parent->blkno, myoff, me->parent->isRoot); + meDelete = TRUE; } } @@ -520,8 +661,14 @@ ginVacuumEntryPage(GinVacuumState *gvs, Buffer buffer, BlockNumber *roots, uint3 * if we already create temporary page, we will make changes in * place */ - ItemPointerData *cleaned = (tmppage == origpage) ? NULL : GinGetPosting(itup); - uint32 newN = ginVacuumPostingList(gvs, GinGetPosting(itup), GinGetNPosting(itup), &cleaned); + Size cleanedSize; + Pointer cleaned = NULL; + uint32 newN = + ginVacuumPostingList(gvs, + GinGetPosting(itup), GinGetNPosting(itup), NULL, + &cleaned, + IndexTupleSize(itup) - GinGetPostingOffset(itup), + &cleanedSize); if (GinGetNPosting(itup) != newN) { @@ -542,23 +689,16 @@ ginVacuumEntryPage(GinVacuumState *gvs, Buffer buffer, BlockNumber *roots, uint3 */ tmppage = PageGetTempPageCopy(origpage); - if (newN > 0) - { - Size pos = ((char *) GinGetPosting(itup)) - ((char *) origpage); - - memcpy(tmppage + pos, cleaned, sizeof(ItemPointerData) * newN); - } - - pfree(cleaned); - /* set itup pointer to new page */ itup = (IndexTuple) PageGetItem(tmppage, PageGetItemId(tmppage, i)); } attnum = gintuple_get_attrnum(&gvs->ginstate, itup); key = gintuple_get_key(&gvs->ginstate, itup, &category); + /* FIXME */ itup = GinFormTuple(&gvs->ginstate, attnum, key, category, - GinGetPosting(itup), newN, true); + cleaned, cleanedSize, newN, true); + pfree(cleaned); PageIndexTupleDelete(tmppage, i); if (PageAddItem(tmppage, (Item) itup, IndexTupleSize(itup), i, false, false) != i) @@ -662,7 +802,7 @@ ginbulkdelete(PG_FUNCTION_ARGS) START_CRIT_SECTION(); PageRestoreTempPage(resPage, page); MarkBufferDirty(buffer); - xlogVacuumPage(gvs.index, buffer); + xlogVacuumPage(gvs.index, buffer, GinPageGetOpaque(page)->u.maxoff); UnlockReleaseBuffer(buffer); END_CRIT_SECTION(); } diff --git a/src/backend/access/gin/ginxlog.c b/src/backend/access/gin/ginxlog.c index 5daabb0..a520cbf 100644 --- a/src/backend/access/gin/ginxlog.c +++ b/src/backend/access/gin/ginxlog.c @@ -106,9 +106,12 @@ static void ginRedoCreatePTree(XLogRecPtr lsn, XLogRecord *record) { ginxlogCreatePostingTree *data = (ginxlogCreatePostingTree *) XLogRecGetData(record); - ItemPointerData *items = (ItemPointerData *) (XLogRecGetData(record) + sizeof(ginxlogCreatePostingTree)); + Pointer ptr = XLogRecGetData(record) + sizeof(ginxlogCreatePostingTree), tmp; Buffer buffer; Page page; + GinState ginstate; + ItemPointerData iptr = {{0, 0}, 0}; + OffsetNumber i; /* Backup blocks are not used in create_ptree records */ Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK)); @@ -118,11 +121,19 @@ ginRedoCreatePTree(XLogRecPtr lsn, XLogRecord *record) page = (Page) BufferGetPage(buffer); GinInitBuffer(buffer, GIN_DATA | GIN_LEAF); - memcpy(GinDataPageGetData(page), items, sizeof(ItemPointerData) * data->nitem); - GinPageGetOpaque(page)->maxoff = data->nitem; + + tmp = ptr; + for (i = 1; i <= data->nitem; i++) + tmp = ginDataPageLeafReadItemPointer(tmp, &iptr); + + memcpy(GinDataPageGetData(page), ptr, tmp - ptr); + + GinPageGetOpaque(page)->u.endoffset = tmp - ptr; PageSetLSN(page, lsn); + updateItemIndexes(page, &ginstate); + MarkBufferDirty(buffer); UnlockReleaseBuffer(buffer); } @@ -182,14 +193,48 @@ ginRedoInsert(XLogRecPtr lsn, XLogRecord *record) if (data->isLeaf) { + ItemPointer startIptr = (ItemPointer) (XLogRecGetData(record) + sizeof(ginxlogInsert)); OffsetNumber i; - ItemPointerData *items = (ItemPointerData *) (XLogRecGetData(record) + sizeof(ginxlogInsert)); + OffsetNumber j; + Pointer dataPtr = (Pointer)(startIptr + 1); + GinState ginstate; + ItemPointerData iptr = {{0, 0}, 0}, prev_iptr; + char pageCopy[BLCKSZ]; + Pointer ptr, destPtr, dataFinish; + Pointer endPtr; Assert(GinPageIsLeaf(page)); Assert(data->updateBlkno == InvalidBlockNumber); - for (i = 0; i < data->nitem; i++) - GinDataPageAddItem(page, items + i, data->offset + i); + memcpy(pageCopy, page, BLCKSZ); + ptr = GinDataPageGetData(page); + for (i = 1; i < data->offset ; i++) + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + + ptr = GinDataPageGetData(pageCopy); + for (i = 1; i < data->offset ; i++) + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + + prev_iptr = iptr; + destPtr = page + (ptr - pageCopy); + + dataFinish = dataPtr; + for (j = 0; j < data->nitem; j++) + dataFinish = ginDataPageLeafReadItemPointer(dataFinish, &prev_iptr); + + memcpy(destPtr, dataPtr, dataFinish - dataPtr); + destPtr += dataFinish - dataPtr; + + endPtr = pageCopy + GinPageGetOpaque(pageCopy)->u.endoffset; + while (ptr < endPtr) + { + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + destPtr = ginDataPageLeafWriteItemPointer(destPtr, &iptr, &prev_iptr); + prev_iptr = iptr; + } + + GinPageGetOpaque(page)->u.endoffset = destPtr - page; + updateItemIndexes(page, &ginstate); } else { @@ -206,7 +251,7 @@ ginRedoInsert(XLogRecPtr lsn, XLogRecord *record) pitem = (PostingItem *) (XLogRecGetData(record) + sizeof(ginxlogInsert)); - GinDataPageAddItem(page, pitem, data->offset); + GinPageAddPostingItem(page, pitem, data->offset); } } else @@ -255,6 +300,7 @@ ginRedoSplit(XLogRecPtr lsn, XLogRecord *record) Page lpage, rpage; uint32 flags = 0; + GinState ginstate; if (data->isLeaf) flags |= GIN_LEAF; @@ -280,31 +326,62 @@ ginRedoSplit(XLogRecPtr lsn, XLogRecord *record) if (data->isData) { char *ptr = XLogRecGetData(record) + sizeof(ginxlogSplit); - Size sizeofitem = GinSizeOfDataPageItem(lpage); OffsetNumber i; ItemPointer bound; - for (i = 0; i < data->separator; i++) + if (data->isLeaf) { - GinDataPageAddItem(lpage, ptr, InvalidOffsetNumber); - ptr += sizeofitem; + Pointer tmp, ptr2; + ItemPointerData iptr = {{0, 0}, 0}; + Size lsize, rsize; + + tmp = ptr; + for (i = 1; i <= data->separator; i++) + tmp = ginDataPageLeafReadItemPointer(tmp, &iptr); + lsize = tmp - ptr; + ptr2 = ptr + MAXALIGN(lsize); + tmp = ptr2; + for (; i <= data->nitem; i++) + tmp = ginDataPageLeafReadItemPointer(tmp, &iptr); + rsize = tmp - ptr2; + + Assert(lsize < GinDataPageSize); + Assert(rsize < GinDataPageSize); + + memcpy(GinDataPageGetData(lpage), ptr, lsize); + memcpy(GinDataPageGetData(rpage), ptr2, rsize); + + GinPageGetOpaque(lpage)->u.endoffset = ptr + lsize - lpage; + GinPageGetOpaque(rpage)->u.endoffset = ptr2 + rsize - rpage; + *GinDataPageGetRightBound(lpage) = updateItemIndexes(lpage, &ginstate); + updateItemIndexes(rpage, &ginstate); + + *GinDataPageGetRightBound(rpage) = data->rightbound; + + Assert(GinPageGetOpaque(lpage)->flags == flags); + Assert(GinPageGetOpaque(rpage)->flags == flags); } - - for (i = data->separator; i < data->nitem; i++) + else { - GinDataPageAddItem(rpage, ptr, InvalidOffsetNumber); - ptr += sizeofitem; - } + PostingItem *pitem = (PostingItem *) ptr; + for (i = 0; i < data->separator; i++) + { + GinPageAddPostingItem(lpage, pitem, InvalidOffsetNumber); + ptr++; + } - /* set up right key */ - bound = GinDataPageGetRightBound(lpage); - if (data->isLeaf) - *bound = *(ItemPointerData *) GinDataPageGetItem(lpage, GinPageGetOpaque(lpage)->maxoff); - else - *bound = ((PostingItem *) GinDataPageGetItem(lpage, GinPageGetOpaque(lpage)->maxoff))->key; + for (i = data->separator; i < data->nitem; i++) + { + GinPageAddPostingItem(rpage, pitem, InvalidOffsetNumber); + ptr++; + } + /* set up right key */ + bound = GinDataPageGetRightBound(lpage); + *bound = ((PostingItem *) GinDataPageGetItem(lpage, GinPageGetOpaque(lpage)->u.maxoff))->key; - bound = GinDataPageGetRightBound(rpage); - *bound = data->rightbound; + bound = GinDataPageGetRightBound(rpage); + *bound = data->rightbound; + } } else { @@ -390,10 +467,30 @@ ginRedoVacuumPage(XLogRecPtr lsn, XLogRecord *record) { if (GinPageIsData(page)) { - memcpy(GinDataPageGetData(page), - XLogRecGetData(record) + sizeof(ginxlogVacuumPage), - data->nitem * GinSizeOfDataPageItem(page)); - GinPageGetOpaque(page)->maxoff = data->nitem; + if (GinPageIsLeaf(page)) + { + GinState ginstate; + ItemPointerData iptr = {{0, 0}, 0}; + Pointer ptr, tmp; + OffsetNumber i; + + ptr = XLogRecGetData(record) + sizeof(ginxlogVacuumPage); + tmp = ptr; + for (i = 1; i <= data->nitem; i++) + tmp = ginDataPageLeafReadItemPointer(tmp, &iptr); + + memcpy(GinDataPageGetData(page), ptr, tmp - ptr); + + GinPageGetOpaque(page)->u.endoffset = tmp - page; + updateItemIndexes(page, &ginstate); + } + else + { + memcpy(GinDataPageGetData(page), + XLogRecGetData(record) + sizeof(ginxlogVacuumPage), + data->nitem * sizeof(ItemPointerData)); + GinPageGetOpaque(page)->u.maxoff = data->nitem; + } } else { @@ -554,7 +651,7 @@ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record) /* * Increase counter of heap tuples */ - GinPageGetOpaque(page)->maxoff++; + GinPageGetOpaque(page)->u.maxoff++; PageSetLSN(page, lsn); MarkBufferDirty(buffer); @@ -621,11 +718,11 @@ ginRedoInsertListPage(XLogRecPtr lsn, XLogRecord *record) { /* tail of sublist */ GinPageSetFullRow(page); - GinPageGetOpaque(page)->maxoff = 1; + GinPageGetOpaque(page)->u.maxoff = 1; } else { - GinPageGetOpaque(page)->maxoff = 0; + GinPageGetOpaque(page)->u.maxoff = 0; } for (i = 0; i < data->ntuples; i++) @@ -802,12 +899,7 @@ ginContinueSplit(ginIncompleteSplit *split) ginPrepareDataScan(&btree, reln); PostingItemSetBlockNumber(&(btree.pitem), split->leftBlkno); - if (GinPageIsLeaf(page)) - btree.pitem.key = *(ItemPointerData *) GinDataPageGetItem(page, - GinPageGetOpaque(page)->maxoff); - else - btree.pitem.key = ((PostingItem *) GinDataPageGetItem(page, - GinPageGetOpaque(page)->maxoff))->key; + btree.pitem.key = *GinDataPageGetRightBound(page); } btree.rightblkno = split->rightBlkno; diff --git a/src/include/access/gin_private.h b/src/include/access/gin_private.h index c603521..92cdf0b 100644 --- a/src/include/access/gin_private.h +++ b/src/include/access/gin_private.h @@ -32,11 +32,14 @@ typedef struct GinPageOpaqueData { BlockNumber rightlink; /* next page if any */ - OffsetNumber maxoff; /* number entries on GIN_DATA page: number of - * heap ItemPointers on GIN_DATA|GIN_LEAF page - * or number of PostingItems on GIN_DATA & - * ~GIN_LEAF page. On GIN_LIST page, number of - * heap tuples. */ + union + { + OffsetNumber maxoff; /* number entries on GIN_DATA page: number of + * PostingItems on GIN_DATA & ~GIN_LEAF page. + * On GIN_LIST page, number of heap tuples. */ + uint16 endoffset; /* beginning of free space on a + * GIN_DATA|GIN_LEAF page */ + } u; uint16 flags; /* see bit definitions below */ } GinPageOpaqueData; @@ -196,10 +199,16 @@ typedef signed char GinNullCategory; #define GinCategoryOffset(itup,ginstate) \ (IndexInfoFindDataOffset((itup)->t_info) + \ ((ginstate)->oneCol ? 0 : sizeof(int16))) -#define GinGetNullCategory(itup,ginstate) \ +/*#define GinGetNullCategory(itup,ginstate) \ (*((GinNullCategory *) ((char*)(itup) + GinCategoryOffset(itup,ginstate)))) #define GinSetNullCategory(itup,ginstate,c) \ - (*((GinNullCategory *) ((char*)(itup) + GinCategoryOffset(itup,ginstate))) = (c)) + (*((GinNullCategory *) ((char*)(itup) + GinCategoryOffset(itup,ginstate))) = (c))*/ + +#define GinGetNullCategory(itup,ginstate) \ + (*((GinNullCategory *) ((char*)(itup) + IndexTupleSize(itup) - sizeof(GinNullCategory)))) +#define GinSetNullCategory(itup,ginstate,c) \ + (*((GinNullCategory *) ((char*)(itup) + IndexTupleSize(itup) - sizeof(GinNullCategory))) = (c)) + /* * Access macros for leaf-page entry tuples (see discussion in README) @@ -213,11 +222,11 @@ typedef signed char GinNullCategory; #define GinGetPostingOffset(itup) GinItemPointerGetBlockNumber(&(itup)->t_tid) #define GinSetPostingOffset(itup,n) ItemPointerSetBlockNumber(&(itup)->t_tid,n) -#define GinGetPosting(itup) ((ItemPointer) ((char*)(itup) + GinGetPostingOffset(itup))) +#define GinGetPosting(itup) ((Pointer) ((char*)(itup) + GinGetPostingOffset(itup))) #define GinMaxItemSize \ MAXALIGN_DOWN(((BLCKSZ - SizeOfPageHeaderData - \ - MAXALIGN(sizeof(GinPageOpaqueData))) / 3 - sizeof(ItemIdData))) + MAXALIGN(sizeof(GinPageOpaqueData))) / 6 - sizeof(ItemIdData))) /* * Access macros for non-leaf entry tuples @@ -232,16 +241,19 @@ typedef signed char GinNullCategory; #define GinDataPageGetRightBound(page) ((ItemPointer) PageGetContents(page)) #define GinDataPageGetData(page) \ (PageGetContents(page) + MAXALIGN(sizeof(ItemPointerData))) -#define GinSizeOfDataPageItem(page) \ - (GinPageIsLeaf(page) ? sizeof(ItemPointerData) : sizeof(PostingItem)) #define GinDataPageGetItem(page,i) \ - (GinDataPageGetData(page) + ((i)-1) * GinSizeOfDataPageItem(page)) + ((PostingItem *)(GinDataPageGetData(page) + ((i)-1) * sizeof(PostingItem))) -#define GinDataPageGetFreeSpace(page) \ +#define GinNonLeafDataPageGetFreeSpace(page) \ (BLCKSZ - MAXALIGN(SizeOfPageHeaderData) \ - MAXALIGN(sizeof(ItemPointerData)) \ - - GinPageGetOpaque(page)->maxoff * GinSizeOfDataPageItem(page) \ + - GinPageGetOpaque(page)->u.maxoff * sizeof(PostingItem) \ - MAXALIGN(sizeof(GinPageOpaqueData))) +#define GinLeafDataPageGetFreeSpace(page) \ + (BLCKSZ \ + - GinPageGetOpaque(page)->u.endoffset \ + - MAXALIGN(sizeof(GinPageOpaqueData)) \ + - MAXALIGN(sizeof(GinDataLeafItemIndex) * GinDataLeafIndexCount)) #define GinMaxLeafDataItems \ ((BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - \ @@ -255,6 +267,29 @@ typedef signed char GinNullCategory; #define GinListPageSize \ ( BLCKSZ - SizeOfPageHeaderData - MAXALIGN(sizeof(GinPageOpaqueData)) ) +typedef struct +{ + ItemPointerData iptr; + OffsetNumber offsetNumber; + uint16 pageOffset; +} GinDataLeafItemIndex; + +#define GinDataLeafIndexCount 32 + +#define GinDataPageSize \ + (BLCKSZ - MAXALIGN(SizeOfPageHeaderData) \ + - MAXALIGN(sizeof(ItemPointerData)) \ + - MAXALIGN(sizeof(GinPageOpaqueData)) \ + - MAXALIGN(sizeof(GinDataLeafItemIndex) * GinDataLeafIndexCount)) + +#define GinDataPageFreeSpacePre(page,ptr) \ + (GinDataPageSize \ + - ((ptr) - GinDataPageGetData(page))) + +#define GinPageGetIndexes(page) \ + ((GinDataLeafItemIndex *)(GinDataPageGetData(page) + GinDataPageSize)) + + /* * Storage type for GIN's reloptions */ @@ -519,22 +554,24 @@ extern void ginInsertValue(GinBtree btree, GinBtreeStack *stack, extern void ginFindParents(GinBtree btree, GinBtreeStack *stack, BlockNumber rootBlkno); /* ginentrypage.c */ -extern IndexTuple GinFormTuple(GinState *ginstate, - OffsetNumber attnum, Datum key, GinNullCategory category, - ItemPointerData *ipd, uint32 nipd, bool errorTooBig); -extern void GinShortenTuple(IndexTuple itup, uint32 nipd); extern void ginPrepareEntryScan(GinBtree btree, OffsetNumber attnum, Datum key, GinNullCategory category, GinState *ginstate); extern void ginEntryFillRoot(GinBtree btree, Buffer root, Buffer lbuf, Buffer rbuf); extern IndexTuple ginPageGetLinkItup(Buffer buf); +extern void ginReadTuple(GinState *ginstate, OffsetNumber attnum, + IndexTuple itup, ItemPointerData *ipd); +extern ItemPointerData updateItemIndexes(Page page, GinState *ginstate); /* gindatapage.c */ +extern char *ginDataPageLeafWriteItemPointer(char *ptr, ItemPointer iptr, ItemPointer prev); +extern Size ginCheckPlaceToDataPageLeaf(ItemPointer iptr, ItemPointer prev, + Size size); extern uint32 ginMergeItemPointers(ItemPointerData *dst, ItemPointerData *a, uint32 na, ItemPointerData *b, uint32 nb); -extern void GinDataPageAddItem(Page page, void *data, OffsetNumber offset); +extern void GinPageAddPostingItem(Page page, PostingItem *data, OffsetNumber offset); extern void GinPageDeletePostingItem(Page page, OffsetNumber offset); typedef struct @@ -724,6 +761,53 @@ extern void ginInsertCleanup(GinState *ginstate, bool vac_delay, IndexBulkDeleteResult *stats); /* + * Function for reading packed ItemPointers. Used in various .c files and + * have to be inline for being fast. + * + * Read next item pointer from leaf data page. Replaces current item pointer + * with the next one. Zero item pointer should be passed in order to read the + * first item pointer. + */ +static inline char * +ginDataPageLeafReadItemPointer(char *ptr, ItemPointer iptr) +{ + uint32 blockNumberIncr; + uint16 offset; + int i; + uint8 v; + + i = 0; + blockNumberIncr = 0; + do + { + v = *ptr; + ptr++; + blockNumberIncr |= (uint32) (v & (~HIGHBIT)) << i; + i += 7; + } + while (IS_HIGHBIT_SET(v)); + + blockNumberIncr += iptr->ip_blkid.bi_lo + (iptr->ip_blkid.bi_hi << 16); + + iptr->ip_blkid.bi_lo = blockNumberIncr & 0xFFFF; + iptr->ip_blkid.bi_hi = (blockNumberIncr >> 16) & 0xFFFF; + + i = 0; + offset = 0; + do + { + v = *ptr; + ptr++; + offset |= (uint32) (v & (~HIGHBIT)) << i; + i += 7; + } while(IS_HIGHBIT_SET(v)); + + iptr->ip_posid = offset; + + return ptr; +} + +/* * Merging the results of several gin scans compares item pointers a lot, * so we want this to be inlined. But if the compiler doesn't support that, * fall back on the non-inline version from itemptr.c. See STATIC_IF_INLINE in