Skip to content
Find file
Fetching contributors…
Cannot retrieve contributors at this time
736 lines (619 sloc) 18 KB
/*-------------------------------------------------------------------------
*
* ginentrypage.c
* page utilities routines for the postgres inverted index access method.
*
*
* Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
* src/backend/access/gin/ginentrypage.c
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/gin_private.h"
#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);
}
/*
* Form a non-leaf entry tuple by copying the key data from the given tuple,
* which can be either a leaf or non-leaf entry tuple.
*
* Any posting list in the source tuple is not copied. The specified child
* block number is inserted into t_tid.
*/
static IndexTuple
GinFormInteriorTuple(IndexTuple itup, Page page, BlockNumber childblk)
{
IndexTuple nitup;
if (GinPageIsLeaf(page) && !GinIsPostingTree(itup))
{
/* Tuple contains a posting list, just copy stuff before that */
uint32 origsize = GinGetPostingOffset(itup);
origsize = MAXALIGN(origsize);
nitup = (IndexTuple) palloc(origsize);
memcpy(nitup, itup, origsize);
/* ... be sure to fix the size header field ... */
nitup->t_info &= ~INDEX_SIZE_MASK;
nitup->t_info |= origsize;
}
else
{
/* Copy the tuple as-is */
nitup = (IndexTuple) palloc(IndexTupleSize(itup));
memcpy(nitup, itup, IndexTupleSize(itup));
}
/* Now insert the correct downlink */
GinSetDownlink(nitup, childblk);
return nitup;
}
/*
* Entry tree is a "static", ie tuple never deletes from it,
* so we don't use right bound, we use rightmost key instead.
*/
static IndexTuple
getRightMostTuple(Page page)
{
OffsetNumber maxoff = PageGetMaxOffsetNumber(page);
return (IndexTuple) PageGetItem(page, PageGetItemId(page, maxoff));
}
static bool
entryIsMoveRight(GinBtree btree, Page page)
{
IndexTuple itup;
OffsetNumber attnum;
Datum key;
GinNullCategory category;
if (GinPageRightMost(page))
return FALSE;
itup = getRightMostTuple(page);
attnum = gintuple_get_attrnum(btree->ginstate, itup);
key = gintuple_get_key(btree->ginstate, itup, &category);
if (ginCompareAttEntries(btree->ginstate,
btree->entryAttnum, btree->entryKey, btree->entryCategory,
attnum, key, category) > 0)
return TRUE;
return FALSE;
}
/*
* Find correct tuple in non-leaf page. It supposed that
* page correctly chosen and searching value SHOULD be on page
*/
static BlockNumber
entryLocateEntry(GinBtree btree, GinBtreeStack *stack)
{
OffsetNumber low,
high,
maxoff;
IndexTuple itup = NULL;
int result;
Page page = BufferGetPage(stack->buffer);
Assert(!GinPageIsLeaf(page));
Assert(!GinPageIsData(page));
if (btree->fullScan)
{
stack->off = FirstOffsetNumber;
stack->predictNumber *= PageGetMaxOffsetNumber(page);
return btree->getLeftMostPage(btree, page);
}
low = FirstOffsetNumber;
maxoff = high = PageGetMaxOffsetNumber(page);
Assert(high >= low);
high++;
while (high > low)
{
OffsetNumber mid = low + ((high - low) / 2);
if (mid == maxoff && GinPageRightMost(page))
{
/* Right infinity */
result = -1;
}
else
{
OffsetNumber attnum;
Datum key;
GinNullCategory category;
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, mid));
attnum = gintuple_get_attrnum(btree->ginstate, itup);
key = gintuple_get_key(btree->ginstate, itup, &category);
result = ginCompareAttEntries(btree->ginstate,
btree->entryAttnum,
btree->entryKey,
btree->entryCategory,
attnum, key, category);
}
if (result == 0)
{
stack->off = mid;
Assert(GinGetDownlink(itup) != GIN_ROOT_BLKNO);
return GinGetDownlink(itup);
}
else if (result > 0)
low = mid + 1;
else
high = mid;
}
Assert(high >= FirstOffsetNumber && high <= maxoff);
stack->off = high;
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, high));
Assert(GinGetDownlink(itup) != GIN_ROOT_BLKNO);
return GinGetDownlink(itup);
}
/*
* Searches correct position for value on leaf page.
* Page should be correctly chosen.
* Returns true if value found on page.
*/
static bool
entryLocateLeafEntry(GinBtree btree, GinBtreeStack *stack)
{
Page page = BufferGetPage(stack->buffer);
OffsetNumber low,
high;
Assert(GinPageIsLeaf(page));
Assert(!GinPageIsData(page));
if (btree->fullScan)
{
stack->off = FirstOffsetNumber;
return TRUE;
}
low = FirstOffsetNumber;
high = PageGetMaxOffsetNumber(page);
if (high < low)
{
stack->off = FirstOffsetNumber;
return false;
}
high++;
while (high > low)
{
OffsetNumber mid = low + ((high - low) / 2);
IndexTuple itup;
OffsetNumber attnum;
Datum key;
GinNullCategory category;
int result;
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, mid));
attnum = gintuple_get_attrnum(btree->ginstate, itup);
key = gintuple_get_key(btree->ginstate, itup, &category);
result = ginCompareAttEntries(btree->ginstate,
btree->entryAttnum,
btree->entryKey,
btree->entryCategory,
attnum, key, category);
if (result == 0)
{
stack->off = mid;
return true;
}
else if (result > 0)
low = mid + 1;
else
high = mid;
}
stack->off = high;
return false;
}
static OffsetNumber
entryFindChildPtr(GinBtree btree, Page page, BlockNumber blkno, OffsetNumber storedOff)
{
OffsetNumber i,
maxoff = PageGetMaxOffsetNumber(page);
IndexTuple itup;
Assert(!GinPageIsLeaf(page));
Assert(!GinPageIsData(page));
/* if page isn't changed, we returns storedOff */
if (storedOff >= FirstOffsetNumber && storedOff <= maxoff)
{
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, storedOff));
if (GinGetDownlink(itup) == blkno)
return storedOff;
/*
* we hope, that needed pointer goes to right. It's true if there
* wasn't a deletion
*/
for (i = storedOff + 1; i <= maxoff; i++)
{
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, i));
if (GinGetDownlink(itup) == blkno)
return i;
}
maxoff = storedOff - 1;
}
/* last chance */
for (i = FirstOffsetNumber; i <= maxoff; i++)
{
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, i));
if (GinGetDownlink(itup) == blkno)
return i;
}
return InvalidOffsetNumber;
}
static BlockNumber
entryGetLeftMostPage(GinBtree btree, Page page)
{
IndexTuple itup;
Assert(!GinPageIsLeaf(page));
Assert(!GinPageIsData(page));
Assert(PageGetMaxOffsetNumber(page) >= FirstOffsetNumber);
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, FirstOffsetNumber));
return GinGetDownlink(itup);
}
static bool
entryIsEnoughSpace(GinBtree btree, Buffer buf, OffsetNumber off)
{
Size itupsz = 0;
Page page = BufferGetPage(buf);
Assert(btree->entry);
Assert(!GinPageIsData(page));
if (btree->isDelete)
{
IndexTuple itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, off));
itupsz = MAXALIGN(IndexTupleSize(itup)) + sizeof(ItemIdData);
}
if (PageGetFreeSpace(page) + itupsz >= MAXALIGN(IndexTupleSize(btree->entry)) + sizeof(ItemIdData))
return true;
return false;
}
/*
* Delete tuple on leaf page if tuples existed and we
* should update it, update old child blkno to new right page
* if child split occurred
*/
static BlockNumber
entryPreparePage(GinBtree btree, Page page, OffsetNumber off)
{
BlockNumber ret = InvalidBlockNumber;
Assert(btree->entry);
Assert(!GinPageIsData(page));
if (btree->isDelete)
{
Assert(GinPageIsLeaf(page));
PageIndexTupleDelete(page, off);
}
if (!GinPageIsLeaf(page) && btree->rightblkno != InvalidBlockNumber)
{
IndexTuple itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, off));
GinSetDownlink(itup, btree->rightblkno);
ret = btree->rightblkno;
}
btree->rightblkno = InvalidBlockNumber;
return ret;
}
/*
* Place tuple on page and fills WAL record
*/
static void
entryPlaceToPage(GinBtree btree, Buffer buf, OffsetNumber off, XLogRecData **prdata)
{
Page page = BufferGetPage(buf);
OffsetNumber placed;
int cnt = 0;
/* these must be static so they can be returned to caller */
static XLogRecData rdata[3];
static ginxlogInsert data;
*prdata = rdata;
data.updateBlkno = entryPreparePage(btree, page, off);
placed = PageAddItem(page, (Item) btree->entry, IndexTupleSize(btree->entry), off, false, false);
if (placed != off)
elog(ERROR, "failed to add item to index page in \"%s\"",
RelationGetRelationName(btree->index));
data.node = btree->index->rd_node;
data.blkno = BufferGetBlockNumber(buf);
data.offset = off;
data.nitem = 1;
data.isDelete = btree->isDelete;
data.isData = false;
data.isLeaf = GinPageIsLeaf(page) ? TRUE : FALSE;
/*
* Prevent full page write if child's split occurs. That is needed to
* remove incomplete splits while replaying WAL
*
* data.updateBlkno contains new block number (of newly created right
* page) for recently splited page.
*/
if (data.updateBlkno == InvalidBlockNumber)
{
rdata[0].buffer = buf;
rdata[0].buffer_std = TRUE;
rdata[0].data = NULL;
rdata[0].len = 0;
rdata[0].next = &rdata[1];
cnt++;
}
rdata[cnt].buffer = InvalidBuffer;
rdata[cnt].data = (char *) &data;
rdata[cnt].len = sizeof(ginxlogInsert);
rdata[cnt].next = &rdata[cnt + 1];
cnt++;
rdata[cnt].buffer = InvalidBuffer;
rdata[cnt].data = (char *) btree->entry;
rdata[cnt].len = IndexTupleSize(btree->entry);
rdata[cnt].next = NULL;
btree->entry = NULL;
}
/*
* Place tuple and split page, original buffer(lbuf) leaves untouched,
* returns shadow page of lbuf filled new data.
* Tuples are distributed between pages by equal size on its, not
* an equal number!
*/
static Page
entrySplitPage(GinBtree btree, Buffer lbuf, Buffer rbuf, OffsetNumber off, XLogRecData **prdata)
{
OffsetNumber i,
maxoff,
separator = InvalidOffsetNumber;
Size totalsize = 0;
Size lsize = 0,
size;
char *ptr;
IndexTuple itup,
leftrightmost = NULL;
Page page;
Page lpage = PageGetTempPageCopy(BufferGetPage(lbuf));
Page rpage = BufferGetPage(rbuf);
Size pageSize = PageGetPageSize(lpage);
/* these must be static so they can be returned to caller */
static XLogRecData rdata[2];
static ginxlogSplit data;
static char tupstore[2 * BLCKSZ];
*prdata = rdata;
data.leftChildBlkno = (GinPageIsLeaf(lpage)) ?
InvalidOffsetNumber : GinGetDownlink(btree->entry);
data.updateBlkno = entryPreparePage(btree, lpage, off);
maxoff = PageGetMaxOffsetNumber(lpage);
ptr = tupstore;
for (i = FirstOffsetNumber; i <= maxoff; i++)
{
if (i == off)
{
size = MAXALIGN(IndexTupleSize(btree->entry));
memcpy(ptr, btree->entry, size);
ptr += size;
totalsize += size + sizeof(ItemIdData);
}
itup = (IndexTuple) PageGetItem(lpage, PageGetItemId(lpage, i));
size = MAXALIGN(IndexTupleSize(itup));
memcpy(ptr, itup, size);
ptr += size;
totalsize += size + sizeof(ItemIdData);
}
if (off == maxoff + 1)
{
size = MAXALIGN(IndexTupleSize(btree->entry));
memcpy(ptr, btree->entry, size);
ptr += size;
totalsize += size + sizeof(ItemIdData);
}
GinInitPage(rpage, GinPageGetOpaque(lpage)->flags, pageSize);
GinInitPage(lpage, GinPageGetOpaque(rpage)->flags, pageSize);
ptr = tupstore;
maxoff++;
lsize = 0;
page = lpage;
for (i = FirstOffsetNumber; i <= maxoff; i++)
{
itup = (IndexTuple) ptr;
if (lsize > totalsize / 2)
{
if (separator == InvalidOffsetNumber)
separator = i - 1;
page = rpage;
}
else
{
leftrightmost = itup;
lsize += MAXALIGN(IndexTupleSize(itup)) + sizeof(ItemIdData);
}
if (PageAddItem(page, (Item) itup, IndexTupleSize(itup), InvalidOffsetNumber, false, false) == InvalidOffsetNumber)
elog(ERROR, "failed to add item to index page in \"%s\"",
RelationGetRelationName(btree->index));
ptr += MAXALIGN(IndexTupleSize(itup));
}
btree->entry = GinFormInteriorTuple(leftrightmost, lpage,
BufferGetBlockNumber(lbuf));
btree->rightblkno = BufferGetBlockNumber(rbuf);
data.node = btree->index->rd_node;
data.rootBlkno = InvalidBlockNumber;
data.lblkno = BufferGetBlockNumber(lbuf);
data.rblkno = BufferGetBlockNumber(rbuf);
data.separator = separator;
data.nitem = maxoff;
data.isData = FALSE;
data.isLeaf = GinPageIsLeaf(lpage) ? TRUE : FALSE;
data.isRootSplit = FALSE;
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 = tupstore;
rdata[1].len = MAXALIGN(totalsize);
rdata[1].next = NULL;
return lpage;
}
/*
* return newly allocated rightmost tuple
*/
IndexTuple
ginPageGetLinkItup(Buffer buf)
{
IndexTuple itup,
nitup;
Page page = BufferGetPage(buf);
itup = getRightMostTuple(page);
nitup = GinFormInteriorTuple(itup, page, BufferGetBlockNumber(buf));
return nitup;
}
/*
* Fills new root by rightest values from child.
* Also called from ginxlog, should not use btree
*/
void
ginEntryFillRoot(GinBtree btree, Buffer root, Buffer lbuf, Buffer rbuf)
{
Page page;
IndexTuple itup;
page = BufferGetPage(root);
itup = ginPageGetLinkItup(lbuf);
if (PageAddItem(page, (Item) itup, IndexTupleSize(itup), InvalidOffsetNumber, false, false) == InvalidOffsetNumber)
elog(ERROR, "failed to add item to index root page");
pfree(itup);
itup = ginPageGetLinkItup(rbuf);
if (PageAddItem(page, (Item) itup, IndexTupleSize(itup), InvalidOffsetNumber, false, false) == InvalidOffsetNumber)
elog(ERROR, "failed to add item to index root page");
pfree(itup);
}
/*
* Set up GinBtree for entry page access
*
* Note: during WAL recovery, there may be no valid data in ginstate
* other than a faked-up Relation pointer; the key datum is bogus too.
*/
void
ginPrepareEntryScan(GinBtree btree, OffsetNumber attnum,
Datum key, GinNullCategory category,
GinState *ginstate)
{
memset(btree, 0, sizeof(GinBtreeData));
btree->index = ginstate->index;
btree->ginstate = ginstate;
btree->findChildPage = entryLocateEntry;
btree->isMoveRight = entryIsMoveRight;
btree->findItem = entryLocateLeafEntry;
btree->findChildPtr = entryFindChildPtr;
btree->getLeftMostPage = entryGetLeftMostPage;
btree->isEnoughSpace = entryIsEnoughSpace;
btree->placeToPage = entryPlaceToPage;
btree->splitPage = entrySplitPage;
btree->fillRoot = ginEntryFillRoot;
btree->isData = FALSE;
btree->searchMode = FALSE;
btree->fullScan = FALSE;
btree->isBuild = FALSE;
btree->entryAttnum = attnum;
btree->entryKey = key;
btree->entryCategory = category;
btree->isDelete = FALSE;
}
Something went wrong with that request. Please try again.