*** a/src/backend/access/gin/README --- b/src/backend/access/gin/README *************** *** 145,150 **** none appear in the key entry itself. The separate pages are called a --- 145,151 ---- 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: *** a/src/backend/access/gin/gindatapage.c --- b/src/backend/access/gin/gindatapage.c *************** *** 17,22 **** --- 17,113 ---- #include "access/gin_private.h" #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); + } + int ginCompareItemPointers(ItemPointer a, ItemPointer b) { *************** *** 159,164 **** dataLocateItem(GinBtree btree, GinBtreeStack *stack) --- 250,322 ---- } /* + * 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); + OffsetNumber i, maxoff, first = FirstOffsetNumber; + ItemPointerData iptr = {{0,0},0}; + int cmp; + + maxoff = GinPageGetOpaque(page)->maxoff; + + /* + * At first, search index at the end of page. As the result we narrow + * [first, maxoff] range. + */ + for (i = 0; i < GinDataLeafIndexCount; i++) + { + GinDataLeafItemIndex *index = &GinPageGetIndexes(page)[i]; + if (index->offsetNumer == InvalidOffsetNumber) + break; + + cmp = ginCompareItemPointers(&index->iptr, btree->items + btree->curitem); + if (cmp < 0) + { + ptr = GinDataPageGetData(page) + index->pageOffset; + first = index->offsetNumer; + iptr = index->iptr; + } + else + { + maxoff = index->offsetNumer - 1; + break; + } + } + + /* Search page in [first, maxoff] range found by page index */ + for (i = first; i <= maxoff; i++) + { + *ptrOut = ptr; + *iptrOut = iptr; + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + + cmp = ginCompareItemPointers(btree->items + btree->curitem, &iptr); + if (cmp == 0) + { + *offset = i; + return true; + } + if (cmp < 0) + { + *offset = i; + return false; + } + } + + *ptrOut = ptr; + *iptrOut = iptr; + *offset = GinPageGetOpaque(page)->maxoff + 1; + return false; + } + + + /* * Searches correct position for value on leaf page. * Page should be correctly chosen. * Returns true if value found on page. *************** *** 167,175 **** static bool dataLocateLeafItem(GinBtree btree, GinBtreeStack *stack) { Page page = BufferGetPage(stack->buffer); ! OffsetNumber low, ! high; ! int result; Assert(GinPageIsLeaf(page)); Assert(GinPageIsData(page)); --- 325,332 ---- dataLocateLeafItem(GinBtree btree, GinBtreeStack *stack) { Page page = BufferGetPage(stack->buffer); ! ItemPointerData iptr; ! Pointer ptr; Assert(GinPageIsLeaf(page)); Assert(GinPageIsData(page)); *************** *** 180,215 **** 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; } /* --- 337,344 ---- return TRUE; } ! return findInLeafPage(btree, page, &stack->off, &iptr, &ptr); } /* *************** *** 333,345 **** dataIsEnoughSpace(GinBtree btree, Buffer buf, OffsetNumber off) if (GinPageIsLeaf(page)) { ! if (GinPageRightMost(page) && off > GinPageGetOpaque(page)->maxoff) { ! if ((btree->nitem - btree->curitem) * sizeof(ItemPointerData) <= GinDataPageGetFreeSpace(page)) ! return true; } ! else if (sizeof(ItemPointerData) <= GinDataPageGetFreeSpace(page)) return true; } else if (sizeof(PostingItem) <= GinDataPageGetFreeSpace(page)) return true; --- 462,498 ---- if (GinPageIsLeaf(page)) { ! int i, n, j; ! ItemPointerData iptr = {{0,0},0}; ! Size size = 0; ! ! /* ! * Calculate additional size using worst case assumption: varbyte ! * encoding from zero item pointer. Also use worst case assumption about ! * alignment. ! */ ! n = GinPageGetOpaque(page)->maxoff; ! ! if (GinPageRightMost(page) && off > n) ! { ! for (j = btree->curitem; j < btree->nitem; j++) ! { ! size = ginCheckPlaceToDataPageLeaf(&btree->items[j], ! (j == btree->curitem) ? (&iptr) : &btree->items[j - 1], ! size); ! i++; ! } ! } ! else { ! j = btree->curitem; ! size = ginCheckPlaceToDataPageLeaf(&btree->items[j], &iptr, size); } ! size += MAXIMUM_ALIGNOF; ! ! if (GinDataPageGetFreeSpace(page) >= size) return true; + } else if (sizeof(PostingItem) <= GinDataPageGetFreeSpace(page)) return true; *************** *** 380,391 **** 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; *prdata = rdata; Assert(GinPageIsData(page)); --- 533,544 ---- dataPlaceToPage(GinBtree btree, Buffer buf, OffsetNumber off, XLogRecData **prdata) { Page page = BufferGetPage(buf); 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)); *************** *** 395,401 **** 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.isDelete = FALSE; data.isData = TRUE; data.isLeaf = GinPageIsLeaf(page) ? TRUE : FALSE; --- 548,554 ---- data.node = btree->index->rd_node; data.blkno = BufferGetBlockNumber(buf); data.offset = off; ! data.nitem = 0; data.isDelete = FALSE; data.isData = TRUE; data.isLeaf = GinPageIsLeaf(page) ? TRUE : FALSE; *************** *** 423,457 **** 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) { ! /* usually, create index... */ ! uint32 savedPos = btree->curitem; ! while (btree->curitem < btree->nitem) { ! GinDataPageAddItem(page, btree->items + btree->curitem, off); ! off++; ! btree->curitem++; } - data.nitem = btree->curitem - savedPos; - rdata[cnt].len = sizeofitem * data.nitem; } else { ! GinDataPageAddItem(page, btree->items + btree->curitem, off); btree->curitem++; } } ! else ! GinDataPageAddItem(page, &(btree->pitem), off); } /* --- 576,908 ---- rdata[cnt].next = &rdata[cnt + 1]; cnt++; if (GinPageIsLeaf(page)) { ! int i = 0, j, max_j; ! Pointer ptr = GinDataPageGetData(page), next_ptr, insertStart; ! ItemPointerData iptr = {{0,0},0}, next_iptr; ! char pageCopy[BLCKSZ]; ! int maxoff = GinPageGetOpaque(page)->maxoff; ! int copySize = 0; ! ! /* ! * 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 (off <= maxoff) ! { ! /* ! * 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 - GinDataPageGetFreeSpace(page) - ! (next_ptr - GinDataPageGetData(page)); ! memcpy(pageCopy, next_ptr, copySize); ! } ! ! /* Check how many items we're going to add */ ! if (GinPageRightMost(page) && off > maxoff) ! max_j = btree->nitem; ! else ! max_j = btree->curitem + 1; ! ! /* Place items to the page while we have enough of space */ ! *((ItemPointerData *)insertData) = iptr; ! 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++; ! } ! ! /* 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 (off <= maxoff) { ! ptr = ginDataPageLeafWriteItemPointer(ptr, &next_iptr, &iptr); ! Assert(GinDataPageFreeSpacePre(page,ptr) >= 0); ! memcpy(ptr, pageCopy, copySize); ! } ! GinPageGetOpaque(page)->maxoff += i; ! ! if (GinDataPageFreeSpacePre(page,ptr) < 0) ! elog(ERROR, "Not enough of 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; ! ! GinDataPageAddItem(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)->maxoff = j; \ ! page = rpage; \ ! ptr = GinDataPageGetData(rpage); \ ! 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, ! maxoff; ! Size totalsize = 0, prevTotalsize; ! Pointer ptr, copyPtr; ! 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; ! ! /* 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); ! ! maxoff = GinPageGetOpaque(lpage)->maxoff; ! ! /* 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)->maxoff = 0; ! GinPageGetOpaque(rpage)->maxoff = 0; ! ! /* Calculate the whole size we're going to place */ ! copyPtr = GinDataPageGetData(lpageCopy); ! iptr.ip_blkid.bi_hi = 0; ! iptr.ip_blkid.bi_lo = 0; ! iptr.ip_posid = 0; ! for (i = FirstOffsetNumber; i <= maxoff; i++) ! { ! 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); ! } ! ! if (off == maxoff + 1) ! { ! 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 ! ) { ! maxItemIndex++; ! totalCount++; ! maxItemSize = Max(maxItemSize, newTotalsize - totalsize); ! totalsize = newTotalsize; ! ! prevIptr = iptr; ! if (maxItemIndex < btree->nitem) ! iptr = btree->items[maxItemIndex]; } } else { ! prevTotalsize = totalsize; ! totalsize = ginCheckPlaceToDataPageLeaf(&iptr, &prevIptr, totalsize); ! maxItemIndex++; ! ! totalCount++; ! maxItemSize = Max(maxItemSize, totalsize - prevTotalsize); ! } ! } ! ! /* ! * 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); ! for (i = FirstOffsetNumber; i <= maxoff; i++) ! { ! 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; ! } ! } ! ! copyPtr = ginDataPageLeafReadItemPointer(copyPtr, &iptr); ! ! ptr = ginDataPageLeafWriteItemPointer(ptr, &iptr, &prevIptr); ! Assert(GinDataPageFreeSpacePre(page, ptr) >= 0); ! ! prevIptr = iptr; ! ! CHECK_SWITCH_TO_RPAGE; ! } ! ! if (off == maxoff + 1) ! { ! 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; } } ! ! GinPageGetOpaque(rpage)->maxoff = j - 1; ! ! 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 = GinPageGetOpaque(lpage)->maxoff; ! data.nitem = GinPageGetOpaque(lpage)->maxoff + GinPageGetOpaque(rpage)->maxoff; ! 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 - GinDataPageGetFreeSpace(lpage); ! rdata[1].next = &rdata[2]; ! ! rdata[2].buffer = InvalidBuffer; ! rdata[2].data = GinDataPageGetData(rpage); ! rdata[2].len = GinDataPageSize - GinDataPageGetFreeSpace(rpage); ! rdata[2].next = NULL; ! ! return lpage; } /* *************** *** 461,467 **** dataPlaceToPage(GinBtree btree, Buffer buf, OffsetNumber off, XLogRecData **prda * left page */ static Page ! dataSplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, XLogRecData **prdata) { char *ptr; OffsetNumber separator; --- 912,919 ---- * left page */ static Page ! dataSplitPageInternal(GinBtree btree, Buffer lbuf, Buffer rbuf, ! OffsetNumber off, XLogRecData **prdata) { char *ptr; OffsetNumber separator; *************** *** 563,569 **** 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.isRootSplit = FALSE; data.rightbound = oldbound; --- 1015,1021 ---- data.separator = separator; data.nitem = maxoff; data.isData = TRUE; ! data.isLeaf = FALSE; data.isRootSplit = FALSE; data.rightbound = oldbound; *************** *** 581,586 **** dataSplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, XLogRe --- 1033,1092 ---- } /* + * 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 ptr; + ItemPointerData iptr; + int j = 0, maxoff, i; + + /* Iterate over page */ + + maxoff = GinPageGetOpaque(page)->maxoff; + ptr = GinDataPageGetData(page); + iptr.ip_blkid.bi_lo = 0; + iptr.ip_blkid.bi_hi = 0; + iptr.ip_posid = 0; + + for (i = FirstOffsetNumber; i <= maxoff; i++) + { + /* Place next page index entry if it's time to */ + if (i * (GinDataLeafIndexCount + 1) > (j + 1) * maxoff) + { + GinPageGetIndexes(page)[j].iptr = iptr; + GinPageGetIndexes(page)[j].offsetNumer = i; + GinPageGetIndexes(page)[j].pageOffset = ptr - GinDataPageGetData(page); + j++; + } + ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + } + /* Fill rest of page indexes with InvalidOffsetNumber if any */ + for (; j < GinDataLeafIndexCount; j++) + { + GinPageGetIndexes(page)[j].offsetNumer = InvalidOffsetNumber; + } + return iptr; + } + + /* * Fills new root by right bound values from child. * Also called from ginxlog, should not use btree */ *** a/src/backend/access/gin/ginentrypage.c --- b/src/backend/access/gin/ginentrypage.c *************** *** 18,162 **** #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. */ ! IndexTuple ! GinFormTuple(GinState *ginstate, ! OffsetNumber attnum, Datum key, GinNullCategory category, ! ItemPointerData *ipd, uint32 nipd, ! bool errorTooBig) { ! 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; ! } ! /* ! * 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); } - - /* - * 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); } /* --- 18,41 ---- #include "utils/rel.h" /* ! * Read item pointers from leaf data page. Information is stored in the same ! * manner as in leaf data pages. */ ! void ! ginReadTuple(GinState *ginstate, OffsetNumber attnum, ! IndexTuple itup, ItemPointerData *ipd) { ! Pointer ptr; ! int nipd = GinGetNPosting(itup), i; ! ItemPointerData ip = {{0,0},0}; ! ptr = GinGetPosting(itup); ! for (i = 0; i < nipd; i++) { ! ptr = ginDataPageLeafReadItemPointer(ptr, &ip); ! ipd[i] = ip; } } /* *** a/src/backend/access/gin/ginfast.c --- b/src/backend/access/gin/ginfast.c *************** *** 23,28 **** --- 23,29 ---- #include "miscadmin.h" #include "utils/memutils.h" #include "utils/rel.h" + #include "access/htup_details.h" #define GIN_PAGE_FREESIZE \ *************** *** 437,442 **** ginHeapTupleFastInsert(GinState *ginstate, GinTupleCollector *collector) --- 438,526 ---- 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,493 **** ginHeapTupleFastCollect(GinState *ginstate, { IndexTuple itup; ! itup = GinFormTuple(ginstate, attnum, entries[i], categories[i], ! NULL, 0, true); itup->t_tid = *ht_ctid; collector->tuples[collector->ntuples++] = itup; collector->sumsize += IndexTupleSize(itup); --- 570,576 ---- { IndexTuple itup; ! itup = GinFastFormTuple(ginstate, attnum, entries[i], categories[i]); itup->t_tid = *ht_ctid; collector->tuples[collector->ntuples++] = itup; collector->sumsize += IndexTupleSize(itup); *** a/src/backend/access/gin/ginget.c --- b/src/backend/access/gin/ginget.c *************** *** 73,90 **** findItemInPostingPage(Page page, ItemPointer item, OffsetNumber *off) { OffsetNumber maxoff = GinPageGetOpaque(page)->maxoff; int res; if (GinPageGetOpaque(page)->flags & GIN_DELETED) /* page was deleted by concurrent vacuum */ return false; /* * scan page to find equal or first greater value */ for (*off = FirstOffsetNumber; *off <= maxoff; (*off)++) { ! res = ginCompareItemPointers(item, (ItemPointer) GinDataPageGetItem(page, *off)); if (res <= 0) return true; } --- 73,94 ---- { OffsetNumber maxoff = GinPageGetOpaque(page)->maxoff; int res; + Pointer ptr; + ItemPointerData iptr = {{0, 0}, 0}; if (GinPageGetOpaque(page)->flags & GIN_DELETED) /* page was deleted by concurrent vacuum */ return false; + ptr = GinDataPageGetData(page); /* * scan page to find equal or first greater value */ for (*off = FirstOffsetNumber; *off <= maxoff; (*off)++) { ! ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); + res = ginCompareItemPointers(item, &iptr); if (res <= 0) return true; } *************** *** 148,162 **** scanPostingTree(Relation index, GinScanEntry scanEntry, */ for (;;) { page = BufferGetPage(buffer); if ((GinPageGetOpaque(page)->flags & GIN_DELETED) == 0 && ! GinPageGetOpaque(page)->maxoff >= FirstOffsetNumber) { ! tbm_add_tuples(scanEntry->matchBitmap, ! (ItemPointer) GinDataPageGetItem(page, FirstOffsetNumber), ! GinPageGetOpaque(page)->maxoff, false); ! scanEntry->predictNumberResult += GinPageGetOpaque(page)->maxoff; } if (GinPageRightMost(page)) --- 152,176 ---- */ for (;;) { + OffsetNumber maxoff, i; + page = BufferGetPage(buffer); + maxoff = GinPageGetOpaque(page)->maxoff; if ((GinPageGetOpaque(page)->flags & GIN_DELETED) == 0 && ! maxoff >= FirstOffsetNumber) { ! ItemPointerData iptr = {{0, 0}, 0}; ! Pointer ptr; ! ! ptr = GinDataPageGetData(page); ! for (i = FirstOffsetNumber; i <= maxoff; i++) ! { ! ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); ! tbm_add_tuples(scanEntry->matchBitmap, &iptr, 1, false); ! } ! ! scanEntry->predictNumberResult += maxoff; } if (GinPageRightMost(page)) *************** *** 344,351 **** collectMatchBitmap(GinBtreeData *btree, GinBtreeStack *stack, } else { tbm_add_tuples(scanEntry->matchBitmap, ! GinGetPosting(itup), GinGetNPosting(itup), false); scanEntry->predictNumberResult += GinGetNPosting(itup); } --- 358,369 ---- } else { + ItemPointerData *ipd = (ItemPointerData *)palloc( + sizeof(ItemPointerData) * GinGetNPosting(itup)); + ginReadTuple(btree->ginstate, scanEntry->attnum, itup, ipd); + tbm_add_tuples(scanEntry->matchBitmap, ! ipd, GinGetNPosting(itup), false); scanEntry->predictNumberResult += GinGetNPosting(itup); } *************** *** 438,443 **** restartScanEntry: --- 456,464 ---- BlockNumber rootPostingTree = GinGetPostingTree(itup); GinPostingTreeScan *gdi; Page page; + OffsetNumber maxoff, i; + Pointer ptr; + ItemPointerData iptr = {{0,0},0}; /* * We should unlock entry page before touching posting tree to *************** *** 465,474 **** restartScanEntry: /* * 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)); LockBuffer(entry->buffer, GIN_UNLOCK); freeGinBtreeStack(gdi->stack); --- 486,502 ---- /* * Keep page content in memory to prevent durable page locking */ ! entry->list = (ItemPointerData *) palloc(BLCKSZ * sizeof(ItemPointerData)); ! maxoff = GinPageGetOpaque(page)->maxoff; ! entry->nlist = maxoff; ! ! ptr = GinDataPageGetData(page); ! ! for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i)) ! { ! ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); ! entry->list[i - FirstOffsetNumber] = iptr; ! } LockBuffer(entry->buffer, GIN_UNLOCK); freeGinBtreeStack(gdi->stack); *************** *** 478,485 **** restartScanEntry: else if (GinGetNPosting(itup) > 0) { entry->nlist = GinGetNPosting(itup); entry->list = (ItemPointerData *) palloc(sizeof(ItemPointerData) * entry->nlist); ! memcpy(entry->list, GinGetPosting(itup), sizeof(ItemPointerData) * entry->nlist); entry->isFinished = FALSE; } } --- 506,516 ---- else if (GinGetNPosting(itup) > 0) { entry->nlist = GinGetNPosting(itup); + entry->predictNumberResult = entry->nlist; entry->list = (ItemPointerData *) palloc(sizeof(ItemPointerData) * entry->nlist); ! ! ginReadTuple(ginstate, entry->attnum, itup, entry->list); ! entry->isFinished = FALSE; } } *************** *** 583,594 **** entryGetNextItem(GinState *ginstate, GinScanEntry entry) if (!ItemPointerIsValid(&entry->curItem) || findItemInPostingPage(page, &entry->curItem, &entry->offset)) { /* * Found position equal to or greater than stored */ ! entry->nlist = GinPageGetOpaque(page)->maxoff; ! memcpy(entry->list, GinDataPageGetItem(page, FirstOffsetNumber), ! GinPageGetOpaque(page)->maxoff * sizeof(ItemPointerData)); LockBuffer(entry->buffer, GIN_UNLOCK); --- 614,636 ---- if (!ItemPointerIsValid(&entry->curItem) || findItemInPostingPage(page, &entry->curItem, &entry->offset)) { + OffsetNumber maxoff, i; + Pointer ptr; + ItemPointerData iptr = {{0,0},0}; + /* * Found position equal to or greater than stored */ ! maxoff = GinPageGetOpaque(page)->maxoff; ! entry->nlist = maxoff; ! ! ptr = GinDataPageGetData(page); ! ! for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i)) ! { ! ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); ! entry->list[i - FirstOffsetNumber] = iptr; ! } LockBuffer(entry->buffer, GIN_UNLOCK); *** a/src/backend/access/gin/gininsert.c --- b/src/backend/access/gin/gininsert.c *************** *** 42,55 **** typedef struct * items[] must be in sorted order with no duplicates. */ static BlockNumber ! createPostingTree(Relation index, ItemPointerData *items, uint32 nitems) { BlockNumber blkno; ! Buffer buffer = GinNewBuffer(index); Page page; /* Assert that the items[] array will fit on one page */ - Assert(nitems <= GinMaxLeafDataItems); START_CRIT_SECTION(); --- 42,57 ---- * items[] must be in sorted order with no duplicates. */ static BlockNumber ! createPostingTree(GinState *ginstate, ItemPointerData *items, uint32 nitems) { BlockNumber blkno; ! 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 */ START_CRIT_SECTION(); *************** *** 57,74 **** createPostingTree(Relation index, ItemPointerData *items, uint32 nitems) page = BufferGetPage(buffer); blkno = BufferGetBlockNumber(buffer); - memcpy(GinDataPageGetData(page), items, sizeof(ItemPointerData) * nitems); GinPageGetOpaque(page)->maxoff = nitems; MarkBufferDirty(buffer); ! if (RelationNeedsWAL(index)) { XLogRecPtr recptr; XLogRecData rdata[2]; ginxlogCreatePostingTree data; ! data.node = index->rd_node; data.blkno = blkno; data.nitem = nitems; --- 59,84 ---- page = BufferGetPage(buffer); blkno = BufferGetBlockNumber(buffer); 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); + } + Assert(GinDataPageFreeSpacePre(page, ptr) >= 0); + updateItemIndexes(page, ginstate); MarkBufferDirty(buffer); ! if (RelationNeedsWAL(ginstate->index)) { XLogRecPtr recptr; XLogRecData rdata[2]; ginxlogCreatePostingTree data; ! data.node = ginstate->index->rd_node; data.blkno = blkno; data.nitem = nitems; *************** *** 78,85 **** 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].next = NULL; recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_PTREE, rdata); --- 88,95 ---- rdata[0].next = &rdata[1]; rdata[1].buffer = InvalidBuffer; ! rdata[1].data = GinDataPageGetData(page); ! rdata[1].len = GinDataPageSize - GinDataPageGetFreeSpace(page); rdata[1].next = NULL; recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_PTREE, rdata); *************** *** 93,98 **** createPostingTree(Relation index, ItemPointerData *items, uint32 nitems) --- 103,239 ---- 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,141 **** addItemPointersToLeafTuple(GinState *ginstate, Datum key; GinNullCategory category; IndexTuple res; Assert(!GinIsPostingTree(old)); attnum = gintuple_get_attrnum(ginstate, old); key = gintuple_get_key(ginstate, old, &category); /* 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 { /* posting list would be too big, convert to posting tree */ BlockNumber postingRoot; --- 252,281 ---- 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, ! newItems, newNPosting, false); ! if (!res) { /* posting list would be too big, convert to posting tree */ BlockNumber postingRoot; *************** *** 146,154 **** 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)); /* During index build, count the newly-added data page */ if (buildStats) --- 286,294 ---- * surely small enough to fit on one posting-tree page, and should * already be in order with no duplicates. */ ! postingRoot = createPostingTree(ginstate, ! oldItems, ! oldNPosting); /* During index build, count the newly-added data page */ if (buildStats) *************** *** 194,199 **** buildFreshLeafTuple(GinState *ginstate, --- 334,353 ---- { /* 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,220 **** buildFreshLeafTuple(GinState *ginstate, * Initialize posting tree with as many TIDs as will fit on the first * page. */ ! postingRoot = createPostingTree(ginstate->index, items, ! Min(nitem, GinMaxLeafDataItems)); /* During index build, count the newly-added data page */ if (buildStats) buildStats->nDataPages++; /* Add any remaining TIDs to the posting tree */ ! if (nitem > GinMaxLeafDataItems) { GinPostingTreeScan *gdi; --- 359,374 ---- * Initialize posting tree with as many TIDs as will fit on the first * page. */ ! postingRoot = createPostingTree(ginstate, items, ! itemsCount); /* During index build, count the newly-added data page */ if (buildStats) buildStats->nDataPages++; /* Add any remaining TIDs to the posting tree */ ! if (nitem > itemsCount) { GinPostingTreeScan *gdi; *************** *** 222,229 **** buildFreshLeafTuple(GinState *ginstate, gdi->btree.isBuild = (buildStats != NULL); ginInsertItemPointers(gdi, ! items + GinMaxLeafDataItems, ! nitem - GinMaxLeafDataItems, buildStats); pfree(gdi); --- 376,383 ---- gdi->btree.isBuild = (buildStats != NULL); ginInsertItemPointers(gdi, ! items + itemsCount, ! nitem - itemsCount, buildStats); pfree(gdi); *** a/src/backend/access/gin/ginvacuum.c --- b/src/backend/access/gin/ginvacuum.c *************** *** 41,80 **** typedef struct */ static uint32 ! ginVacuumPostingList(GinVacuumState *gvs, ItemPointerData *items, uint32 nitem, ItemPointerData **cleaned) { uint32 i, j = 0; /* * just scan over ItemPointer array */ for (i = 0; i < nitem; i++) { ! if (gvs->callback(items + i, gvs->callback_state)) { gvs->result->tuples_removed += 1; ! if (!*cleaned) { ! *cleaned = (ItemPointerData *) palloc(sizeof(ItemPointerData) * nitem); if (i != 0) ! memcpy(*cleaned, items, sizeof(ItemPointerData) * i); } } else { gvs->result->num_index_tuples += 1; if (i != j) ! (*cleaned)[j] = items[i]; j++; } } return j; } /* * fills WAL record for vacuum leaf page */ static void --- 41,206 ---- */ static uint32 ! ginVacuumPostingList(GinVacuumState *gvs, Pointer src, uint32 nitem, 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 */ + prevIptr = iptr; for (i = 0; i < nitem; i++) { ! prev = ptr; ! ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); ! if (gvs->callback(&iptr, gvs->callback_state)) { gvs->result->tuples_removed += 1; ! if (!dst) { ! dst = (Pointer) palloc(size); ! *cleaned = dst; if (i != 0) ! { ! memcpy(dst, src, prev - src); ! dst += prev - src; ! } } } else { gvs->result->num_index_tuples += 1; if (i != j) ! 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 *************** *** 101,107 **** xlogVacuumPage(Relation index, Buffer buffer) backup = GinDataPageGetData(page); data.nitem = GinPageGetOpaque(page)->maxoff; if (data.nitem) ! len = MAXALIGN(sizeof(ItemPointerData) * data.nitem); } else { --- 227,233 ---- backup = GinDataPageGetData(page); data.nitem = GinPageGetOpaque(page)->maxoff; if (data.nitem) ! len = MAXALIGN(GinDataPageSize - GinDataPageGetFreeSpace(page)); } else { *************** *** 178,187 **** ginVacuumPostingTreeLeaves(GinVacuumState *gvs, BlockNumber blkno, bool isRoot, { OffsetNumber newMaxOff, oldMaxOff = GinPageGetOpaque(page)->maxoff; ! ItemPointerData *cleaned = NULL; newMaxOff = ginVacuumPostingList(gvs, ! (ItemPointer) GinDataPageGetData(page), oldMaxOff, &cleaned); /* saves changes about deleted tuple ... */ if (oldMaxOff != newMaxOff) --- 304,315 ---- { OffsetNumber newMaxOff, oldMaxOff = GinPageGetOpaque(page)->maxoff; ! Pointer cleaned = NULL; ! Size newSize; newMaxOff = ginVacuumPostingList(gvs, ! GinDataPageGetData(page), oldMaxOff, &cleaned, ! GinDataPageSize - GinDataPageGetFreeSpace(page), &newSize); /* saves changes about deleted tuple ... */ if (oldMaxOff != newMaxOff) *************** *** 189,197 **** ginVacuumPostingTreeLeaves(GinVacuumState *gvs, BlockNumber blkno, bool isRoot, START_CRIT_SECTION(); if (newMaxOff > 0) ! memcpy(GinDataPageGetData(page), cleaned, sizeof(ItemPointerData) * newMaxOff); pfree(cleaned); GinPageGetOpaque(page)->maxoff = newMaxOff; MarkBufferDirty(buffer); xlogVacuumPage(gvs->index, buffer); --- 317,327 ---- START_CRIT_SECTION(); if (newMaxOff > 0) ! memcpy(GinDataPageGetData(page), cleaned, newSize); ! pfree(cleaned); GinPageGetOpaque(page)->maxoff = newMaxOff; + updateItemIndexes(page, &gvs->ginstate); MarkBufferDirty(buffer); xlogVacuumPage(gvs->index, buffer); *************** *** 520,527 **** 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); if (GinGetNPosting(itup) != newN) { --- 650,662 ---- * if we already create temporary page, we will make changes in * place */ ! Size cleanedSize; ! Pointer cleaned = NULL; ! uint32 newN = ! ginVacuumPostingList(gvs, ! GinGetPosting(itup), GinGetNPosting(itup), &cleaned, ! IndexTupleSize(itup) - GinGetPostingOffset(itup), ! &cleanedSize); if (GinGetNPosting(itup) != newN) { *************** *** 542,564 **** 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); itup = GinFormTuple(&gvs->ginstate, attnum, key, category, ! GinGetPosting(itup), newN, true); PageIndexTupleDelete(tmppage, i); if (PageAddItem(tmppage, (Item) itup, IndexTupleSize(itup), i, false, false) != i) --- 677,692 ---- */ tmppage = PageGetTempPageCopy(origpage); /* 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, ! cleaned, cleanedSize, newN, true); ! pfree(cleaned); PageIndexTupleDelete(tmppage, i); if (PageAddItem(tmppage, (Item) itup, IndexTupleSize(itup), i, false, false) != i) *** a/src/backend/access/gin/ginxlog.c --- b/src/backend/access/gin/ginxlog.c *************** *** 106,114 **** static void ginRedoCreatePTree(XLogRecPtr lsn, XLogRecord *record) { ginxlogCreatePostingTree *data = (ginxlogCreatePostingTree *) XLogRecGetData(record); ! ItemPointerData *items = (ItemPointerData *) (XLogRecGetData(record) + sizeof(ginxlogCreatePostingTree)); Buffer buffer; Page page; /* Backup blocks are not used in create_ptree records */ Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK)); --- 106,117 ---- ginRedoCreatePTree(XLogRecPtr lsn, XLogRecord *record) { ginxlogCreatePostingTree *data = (ginxlogCreatePostingTree *) XLogRecGetData(record); ! 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,128 **** 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; PageSetLSN(page, lsn); MarkBufferDirty(buffer); UnlockReleaseBuffer(buffer); } --- 121,139 ---- page = (Page) BufferGetPage(buffer); GinInitBuffer(buffer, GIN_DATA | GIN_LEAF); ! ! tmp = ptr; ! for (i = 1; i <= data->nitem; i++) ! tmp = ginDataPageLeafReadItemPointer(tmp, &iptr); ! ! memcpy(GinDataPageGetData(page), ptr, tmp - ptr); ! GinPageGetOpaque(page)->maxoff = data->nitem; PageSetLSN(page, lsn); + updateItemIndexes(page, &ginstate); + MarkBufferDirty(buffer); UnlockReleaseBuffer(buffer); } *************** *** 182,195 **** ginRedoInsert(XLogRecPtr lsn, XLogRecord *record) if (data->isLeaf) { ! OffsetNumber i; ! ItemPointerData *items = (ItemPointerData *) (XLogRecGetData(record) + sizeof(ginxlogInsert)); Assert(GinPageIsLeaf(page)); Assert(data->updateBlkno == InvalidBlockNumber); ! for (i = 0; i < data->nitem; i++) ! GinDataPageAddItem(page, items + i, data->offset + i); } else { --- 193,241 ---- if (data->isLeaf) { ! ItemPointer startIptr = (ItemPointer) (XLogRecGetData(record) + sizeof(ginxlogInsert)); ! OffsetNumber i, maxoff = GinPageGetOpaque(page)->maxoff, j; ! Pointer dataPtr = (Pointer)(startIptr + 1); ! GinState ginstate; ! ItemPointerData iptr = {{0, 0}, 0}, prev_iptr; ! char pageCopy[BLCKSZ]; ! Pointer ptr, destPtr, dataFinish; Assert(GinPageIsLeaf(page)); Assert(data->updateBlkno == InvalidBlockNumber); ! 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; ! ! for (; i <= maxoff; i++) ! { ! ptr = ginDataPageLeafReadItemPointer(ptr, &iptr); ! destPtr = ginDataPageLeafWriteItemPointer(destPtr, &iptr, &prev_iptr); ! prev_iptr = iptr; ! } ! ! GinPageGetOpaque(page)->maxoff = maxoff + data->nitem; ! updateItemIndexes(page, &ginstate); } else { *************** *** 255,260 **** ginRedoSplit(XLogRecPtr lsn, XLogRecord *record) --- 301,307 ---- Page lpage, rpage; uint32 flags = 0; + GinState ginstate; if (data->isLeaf) flags |= GIN_LEAF; *************** *** 284,310 **** ginRedoSplit(XLogRecPtr lsn, XLogRecord *record) OffsetNumber i; ItemPointer bound; ! for (i = 0; i < data->separator; i++) { ! GinDataPageAddItem(lpage, ptr, InvalidOffsetNumber); ! ptr += sizeofitem; } ! ! for (i = data->separator; i < data->nitem; i++) { ! GinDataPageAddItem(rpage, ptr, InvalidOffsetNumber); ! ptr += sizeofitem; ! } ! /* 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; ! bound = GinDataPageGetRightBound(rpage); ! *bound = data->rightbound; } else { --- 331,391 ---- OffsetNumber i; ItemPointer bound; ! if (data->isLeaf) { ! 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)->maxoff = data->separator; ! GinPageGetOpaque(rpage)->maxoff = data->nitem - data->separator; ! *GinDataPageGetRightBound(lpage) = updateItemIndexes(lpage, &ginstate); ! updateItemIndexes(rpage, &ginstate); ! ! *GinDataPageGetRightBound(rpage) = data->rightbound; ! ! Assert(GinPageGetOpaque(lpage)->flags == flags); ! Assert(GinPageGetOpaque(rpage)->flags == flags); } ! else { ! for (i = 0; i < data->separator; i++) ! { ! GinDataPageAddItem(lpage, ptr, InvalidOffsetNumber); ! ptr += sizeofitem; ! } ! for (i = data->separator; i < data->nitem; i++) ! { ! GinDataPageAddItem(rpage, ptr, InvalidOffsetNumber); ! ptr += sizeofitem; ! } ! /* 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; ! bound = GinDataPageGetRightBound(rpage); ! *bound = data->rightbound; ! } } else { *************** *** 390,399 **** ginRedoVacuumPage(XLogRecPtr lsn, XLogRecord *record) { if (GinPageIsData(page)) { ! memcpy(GinDataPageGetData(page), ! XLogRecGetData(record) + sizeof(ginxlogVacuumPage), ! data->nitem * GinSizeOfDataPageItem(page)); ! GinPageGetOpaque(page)->maxoff = data->nitem; } else { --- 471,500 ---- { if (GinPageIsData(page)) { ! 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)->maxoff = data->nitem; ! updateItemIndexes(page, &ginstate); ! } ! else ! { ! memcpy(GinDataPageGetData(page), ! XLogRecGetData(record) + sizeof(ginxlogVacuumPage), ! data->nitem * GinSizeOfDataPageItem(page)); ! GinPageGetOpaque(page)->maxoff = data->nitem; ! } } else { *************** *** 802,813 **** 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.rightblkno = split->rightBlkno; --- 903,909 ---- ginPrepareDataScan(&btree, reln); PostingItemSetBlockNumber(&(btree.pitem), split->leftBlkno); ! btree.pitem.key = *GinDataPageGetRightBound(page); } btree.rightblkno = split->rightBlkno; *** a/src/include/access/gin_private.h --- b/src/include/access/gin_private.h *************** *** 196,205 **** typedef signed char GinNullCategory; #define GinCategoryOffset(itup,ginstate) \ (IndexInfoFindDataOffset((itup)->t_info) + \ ((ginstate)->oneCol ? 0 : sizeof(int16))) ! #define GinGetNullCategory(itup,ginstate) \ (*((GinNullCategory *) ((char*)(itup) + GinCategoryOffset(itup,ginstate)))) #define GinSetNullCategory(itup,ginstate,c) \ ! (*((GinNullCategory *) ((char*)(itup) + GinCategoryOffset(itup,ginstate))) = (c)) /* * Access macros for leaf-page entry tuples (see discussion in README) --- 196,211 ---- #define GinCategoryOffset(itup,ginstate) \ (IndexInfoFindDataOffset((itup)->t_info) + \ ((ginstate)->oneCol ? 0 : sizeof(int16))) ! /*#define GinGetNullCategory(itup,ginstate) \ (*((GinNullCategory *) ((char*)(itup) + GinCategoryOffset(itup,ginstate)))) #define GinSetNullCategory(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,223 **** 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 GinMaxItemSize \ MAXALIGN_DOWN(((BLCKSZ - SizeOfPageHeaderData - \ ! MAXALIGN(sizeof(GinPageOpaqueData))) / 3 - sizeof(ItemIdData))) /* * Access macros for non-leaf entry tuples --- 219,229 ---- #define GinGetPostingOffset(itup) GinItemPointerGetBlockNumber(&(itup)->t_tid) #define GinSetPostingOffset(itup,n) ItemPointerSetBlockNumber(&(itup)->t_tid,n) ! #define GinGetPosting(itup) ((Pointer) ((char*)(itup) + GinGetPostingOffset(itup))) #define GinMaxItemSize \ MAXALIGN_DOWN(((BLCKSZ - SizeOfPageHeaderData - \ ! MAXALIGN(sizeof(GinPageOpaqueData))) / 6 - sizeof(ItemIdData))) /* * Access macros for non-leaf entry tuples *************** *** 255,260 **** typedef signed char GinNullCategory; --- 261,289 ---- #define GinListPageSize \ ( BLCKSZ - SizeOfPageHeaderData - MAXALIGN(sizeof(GinPageOpaqueData)) ) + typedef struct + { + ItemPointerData iptr; + OffsetNumber offsetNumer; + 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,536 **** 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); /* gindatapage.c */ extern int ginCompareItemPointers(ItemPointer a, ItemPointer b); extern uint32 ginMergeItemPointers(ItemPointerData *dst, ItemPointerData *a, uint32 na, ItemPointerData *b, uint32 nb); --- 548,567 ---- extern void ginFindParents(GinBtree btree, GinBtreeStack *stack, BlockNumber rootBlkno); /* ginentrypage.c */ 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 int ginCompareItemPointers(ItemPointer a, ItemPointer b); + 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); *************** *** 724,727 **** extern void ginHeapTupleFastCollect(GinState *ginstate, --- 755,805 ---- 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; + } + #endif /* GIN_PRIVATE_H */