+ arr[i] = reasloffset[ arr[i] ];
}
/*
* gistSplit -- split a page in the tree.
*/
-static IndexTuple *
+IndexTuple *
gistSplit(Relation r,
Buffer buffer,
IndexTuple *itup, /* contains compressed entry */
int *len,
- PageLayout **dist,
+ SplitedPageLayout **dist,
GISTSTATE *giststate)
{
Page p;
GISTPageOpaque opaque;
GIST_SPLITVEC v;
GistEntryVector *entryvec;
- int i,
+ int i, fakeoffset,
nlen;
+ OffsetNumber *realoffset;
+ IndexTuple *cleaneditup = itup;
+ int lencleaneditup = *len;
p = (Page) BufferGetPage(buffer);
opaque = (GISTPageOpaque) PageGetSpecialPointer(p);
*/
if (BufferGetBlockNumber(buffer) == GIST_ROOT_BLKNO)
{
- leftbuf = ReadBuffer(r, P_NEW);
- GISTInitBuffer(leftbuf, opaque->flags);
+ leftbuf = gistReadBuffer(r, P_NEW);
+ GISTInitBuffer(leftbuf, opaque->flags&F_LEAF);
lbknum = BufferGetBlockNumber(leftbuf);
left = (Page) BufferGetPage(leftbuf);
}
left = (Page) PageGetTempPage(p, sizeof(GISTPageOpaqueData));
}
- rightbuf = ReadBuffer(r, P_NEW);
- GISTInitBuffer(rightbuf, opaque->flags);
+ rightbuf = gistReadBuffer(r, P_NEW);
+ GISTInitBuffer(rightbuf, opaque->flags&F_LEAF);
rbknum = BufferGetBlockNumber(rightbuf);
right = (Page) BufferGetPage(rightbuf);
/* generate the item array */
+ realoffset = palloc((*len + 1) * sizeof(OffsetNumber));
entryvec = palloc(GEVHDRSZ + (*len + 1) * sizeof(GISTENTRY));
entryvec->n = *len + 1;
+ fakeoffset = FirstOffsetNumber;
for (i = 1; i <= *len; i++)
{
Datum datum;
bool IsNull;
+ if (!GistPageIsLeaf(p) && GistTupleIsInvalid( itup[i - 1] )) {
+ entryvec->n--;
+ /* remember position of invalid tuple */
+ realoffset[ entryvec->n ] = i;
+ continue;
+ }
+
datum = index_getattr(itup[i - 1], 1, giststate->tupdesc, &IsNull);
- gistdentryinit(giststate, 0, &(entryvec->vector[i]),
+ gistdentryinit(giststate, 0, &(entryvec->vector[fakeoffset]),
datum, r, p, i,
ATTSIZE(datum, giststate->tupdesc, 1, IsNull),
FALSE, IsNull);
+ realoffset[ fakeoffset ] = i;
+ fakeoffset++;
}
- /*
- * now let the user-defined picksplit function set up the split
- * vector; in entryvec have no null value!!
- */
- FunctionCall2(&giststate->picksplitFn[0],
- PointerGetDatum(entryvec),
- PointerGetDatum(&v));
-
- /* compatibility with old code */
- if (v.spl_left[v.spl_nleft - 1] == InvalidOffsetNumber)
- v.spl_left[v.spl_nleft - 1] = (OffsetNumber) *len;
- if (v.spl_right[v.spl_nright - 1] == InvalidOffsetNumber)
- v.spl_right[v.spl_nright - 1] = (OffsetNumber) *len;
-
- v.spl_lattr[0] = v.spl_ldatum;
- v.spl_rattr[0] = v.spl_rdatum;
- v.spl_lisnull[0] = false;
- v.spl_risnull[0] = false;
-
- /*
- * if index is multikey, then we must to try get smaller bounding box
- * for subkey(s)
- */
- if (r->rd_att->natts > 1)
- {
- int MaxGrpId;
-
- v.spl_idgrp = (int *) palloc0(sizeof(int) * (*len + 1));
- v.spl_grpflag = (char *) palloc0(sizeof(char) * (*len + 1));
- v.spl_ngrp = (int *) palloc(sizeof(int) * (*len + 1));
-
- MaxGrpId = gistfindgroup(giststate, entryvec->vector, &v);
-
- /* form union of sub keys for each page (l,p) */
- gistunionsubkey(r, giststate, itup, &v);
-
- /*
- * if possible, we insert equivalent tuples with control by
- * penalty for a subkey(s)
- */
- if (MaxGrpId > 1)
- gistadjsubkey(r, itup, len, &v, giststate);
+ /*
+ * if it was invalid tuple then we need special processing. If
+ * it's possible, we move all invalid tuples on right page.
+ * We should remember, that union with invalid tuples
+ * is a invalid tuple.
+ */
+ if ( entryvec->n != *len + 1 ) {
+ lencleaneditup = entryvec->n-1;
+ cleaneditup = (IndexTuple*)palloc(lencleaneditup * sizeof(IndexTuple));
+ for(i=1;in;i++)
+ cleaneditup[i-1] = itup[ realoffset[ i ]-1 ];
+
+ if ( gistnospace( left, cleaneditup, lencleaneditup ) ) {
+ /* no space on left to put all good tuples, so picksplit */
+ gistUserPicksplit(r, entryvec, &v, cleaneditup, lencleaneditup, giststate);
+ v.spl_leftvalid = true;
+ v.spl_rightvalid = false;
+ gistToRealOffset( v.spl_left, v.spl_nleft, realoffset );
+ gistToRealOffset( v.spl_right, v.spl_nright, realoffset );
+ } else {
+ /* we can try to store all valid tuples on one page */
+ v.spl_right = (OffsetNumber*)palloc( entryvec->n * sizeof(OffsetNumber) );
+ v.spl_left = (OffsetNumber*)palloc( entryvec->n * sizeof(OffsetNumber) );
+
+ if ( lencleaneditup==0 ) {
+ /* all tuples are invalid, so moves half of its to right */
+ v.spl_leftvalid = v.spl_rightvalid = false;
+ v.spl_nright = 0;
+ v.spl_nleft = 0;
+ for(i=1;i<=*len;i++)
+ if ( i-1<*len/2 )
+ v.spl_left[ v.spl_nleft++ ] = i;
+ else
+ v.spl_right[ v.spl_nright++ ] = i;
+ } else {
+ /* we will not call gistUserPicksplit, just put good
+ tuples on left and invalid on right */
+ v.spl_nleft = lencleaneditup;
+ v.spl_nright = 0;
+ for(i=1;in;i++)
+ v.spl_left[i-1] = i;
+ gistToRealOffset( v.spl_left, v.spl_nleft, realoffset );
+ v.spl_lattr[0] = v.spl_ldatum = (Datum)0;
+ v.spl_rattr[0] = v.spl_rdatum = (Datum)0;
+ v.spl_lisnull[0] = true;
+ v.spl_risnull[0] = true;
+ gistunionsubkey(r, giststate, itup, &v, true);
+ v.spl_leftvalid = true;
+ v.spl_rightvalid = false;
+ }
+ }
+ } else {
+ /* there is no invalid tuples, so usial processing */
+ gistUserPicksplit(r, entryvec, &v, itup, *len, giststate);
+ v.spl_leftvalid = v.spl_rightvalid = true;
}
+
/* form left and right vector */
- lvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * v.spl_nleft);
- rvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * v.spl_nright);
+ lvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * (*len+1));
+ rvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * (*len+1));
for (i = 0; i < v.spl_nleft; i++)
lvectup[i] = itup[v.spl_left[i] - 1];
for (i = 0; i < v.spl_nright; i++)
rvectup[i] = itup[v.spl_right[i] - 1];
+ /* place invalid tuples on right page if itsn't done yet */
+ for (fakeoffset = entryvec->n; fakeoffset < *len+1 && lencleaneditup; fakeoffset++) {
+ rvectup[v.spl_nright++] = itup[realoffset[fakeoffset] - 1];
+ }
/* write on disk (may need another split) */
if (gistnospace(right, rvectup, v.spl_nright))
{
int i;
- PageLayout *d, *origd=*dist;
+ SplitedPageLayout *d, *origd=*dist;
nlen = v.spl_nright;
newtup = gistSplit(r, rightbuf, rvectup, &nlen, dist, giststate);
nlen = 1;
newtup = (IndexTuple *) palloc(sizeof(IndexTuple) * 1);
- newtup[0] = gistFormTuple(giststate, r, v.spl_rattr, v.spl_rattrsize, v.spl_risnull);
- ItemPointerSet(&(newtup[0]->t_tid), rbknum, FirstOffsetNumber);
+ newtup[0] = ( v.spl_rightvalid ) ? gistFormTuple(giststate, r, v.spl_rattr, v.spl_rattrsize, v.spl_risnull)
+ : gist_form_invalid_tuple( rbknum );
+ ItemPointerSetBlockNumber(&(newtup[0]->t_tid), rbknum);
}
if (gistnospace(left, lvectup, v.spl_nleft))
int llen = v.spl_nleft;
IndexTuple *lntup;
int i;
- PageLayout *d, *origd=*dist;
+ SplitedPageLayout *d, *origd=*dist;
lntup = gistSplit(r, leftbuf, lvectup, &llen, dist, giststate);
nlen += 1;
newtup = (IndexTuple *) repalloc(newtup, sizeof(IndexTuple) * nlen);
- newtup[nlen - 1] = gistFormTuple(giststate, r, v.spl_lattr, v.spl_lattrsize, v.spl_lisnull);
- ItemPointerSet(&(newtup[nlen - 1]->t_tid), lbknum, FirstOffsetNumber);
+ newtup[nlen - 1] = ( v.spl_leftvalid ) ? gistFormTuple(giststate, r, v.spl_lattr, v.spl_lattrsize, v.spl_lisnull)
+ : gist_form_invalid_tuple( lbknum );
+ ItemPointerSetBlockNumber(&(newtup[nlen - 1]->t_tid), lbknum);
}
+ GistClearTuplesDeleted(p);
+
*len = nlen;
return newtup;
}
void
-gistnewroot(Relation r, IndexTuple *itup, int len, ItemPointer key, bool xlog_mode)
+gistnewroot(Relation r, IndexTuple *itup, int len, ItemPointer key)
{
Buffer buffer;
Page page;
- buffer = (xlog_mode) ? XLogReadBuffer(false, r, GIST_ROOT_BLKNO) : ReadBuffer(r, GIST_ROOT_BLKNO);
+ buffer = gistReadBuffer(r, GIST_ROOT_BLKNO);
GISTInitBuffer(buffer, 0);
page = BufferGetPage(buffer);
gistfillbuffer(r, page, itup, len, FirstOffsetNumber);
- if ( !xlog_mode && !r->rd_istemp ) {
- gistxlogEntryUpdate xlrec;
+ if ( !r->rd_istemp ) {
XLogRecPtr recptr;
- XLogRecData *rdata = (XLogRecData*)palloc( sizeof(XLogRecData) * ( len + 1 ) );
- int i;
+ XLogRecData *rdata;
- xlrec.node = r->rd_node;
- xlrec.blkno = GIST_ROOT_BLKNO;
- xlrec.todeleteoffnum = InvalidOffsetNumber;
- xlrec.key = *key;
- xlrec.pathlen=0;
-
- rdata[0].buffer = InvalidBuffer;
- rdata[0].data = (char *) &xlrec;
- rdata[0].len = sizeof( gistxlogEntryUpdate );
- rdata[0].next = NULL;
-
- for(i=1; i<=len; i++) {
- rdata[i].buffer = InvalidBuffer;
- rdata[i].data = (char*)(itup[i-1]);
- rdata[i].len = IndexTupleSize(itup[i-1]);
- rdata[i].next = NULL;
- rdata[i-1].next = &(rdata[i]);
- }
+ rdata = formUpdateRdata(r->rd_node, GIST_ROOT_BLKNO,
+ NULL, 0, false, itup, len,
+ key, NULL, 0);
START_CRIT_SECTION();
END_CRIT_SECTION();
}
- if ( xlog_mode )
- LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
WriteBuffer(buffer);
}
-
-/*
- * Bulk deletion of all index entries pointing to a set of heap tuples.
- * The set of target tuples is specified via a callback routine that tells
- * whether any given heap tuple (identified by ItemPointer) is being deleted.
- *
- * Result: a palloc'd struct containing statistical info for VACUUM displays.
- */
-Datum
-gistbulkdelete(PG_FUNCTION_ARGS)
-{
- Relation rel = (Relation) PG_GETARG_POINTER(0);
- IndexBulkDeleteCallback callback = (IndexBulkDeleteCallback) PG_GETARG_POINTER(1);
- void *callback_state = (void *) PG_GETARG_POINTER(2);
- IndexBulkDeleteResult *result;
- BlockNumber num_pages;
- double tuples_removed;
- double num_index_tuples;
- IndexScanDesc iscan;
-
- tuples_removed = 0;
- num_index_tuples = 0;
-
- /*
- * Since GIST is not marked "amconcurrent" in pg_am, caller should
- * have acquired exclusive lock on index relation. We need no locking
- * here.
- */
-
- /*
- * XXX generic implementation --- should be improved!
- */
-
- /* walk through the entire index */
- iscan = index_beginscan(NULL, rel, SnapshotAny, 0, NULL);
- /* including killed tuples */
- iscan->ignore_killed_tuples = false;
-
- while (index_getnext_indexitem(iscan, ForwardScanDirection))
- {
- vacuum_delay_point();
-
- if (callback(&iscan->xs_ctup.t_self, callback_state))
- {
- ItemPointerData indextup = iscan->currentItemData;
- BlockNumber blkno;
- OffsetNumber offnum;
- Buffer buf;
- Page page;
-
- blkno = ItemPointerGetBlockNumber(&indextup);
- offnum = ItemPointerGetOffsetNumber(&indextup);
-
- /* adjust any scans that will be affected by this deletion */
- gistadjscans(rel, GISTOP_DEL, blkno, offnum);
-
- /* delete the index tuple */
- buf = ReadBuffer(rel, blkno);
- page = BufferGetPage(buf);
-
- PageIndexTupleDelete(page, offnum);
- if ( !rel->rd_istemp ) {
- gistxlogEntryUpdate xlrec;
- XLogRecPtr recptr;
- XLogRecData rdata;
-
- xlrec.node = rel->rd_node;
- xlrec.blkno = blkno;
- xlrec.todeleteoffnum = offnum;
- xlrec.pathlen=0;
- ItemPointerSetInvalid( &(xlrec.key) );
-
- rdata.buffer = InvalidBuffer;
- rdata.data = (char *) &xlrec;
- rdata.len = sizeof( gistxlogEntryUpdate );
- rdata.next = NULL;
-
- START_CRIT_SECTION();
-
- recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_ENTRY_DELETE, &rdata);
- PageSetLSN(page, recptr);
- PageSetTLI(page, ThisTimeLineID);
-
- END_CRIT_SECTION();
- }
-
- WriteBuffer(buf);
-
- tuples_removed += 1;
- }
- else
- num_index_tuples += 1;
- }
-
- index_endscan(iscan);
-
- /* return statistics */
- num_pages = RelationGetNumberOfBlocks(rel);
-
- result = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
- result->num_pages = num_pages;
- result->num_index_tuples = num_index_tuples;
- result->tuples_removed = tuples_removed;
-
- PG_RETURN_POINTER(result);
-}
-
void
initGISTstate(GISTSTATE *giststate, Relation index)
{
/* no work */
}
-#ifdef GISTDEBUG
-static void
-gist_dumptree(Relation r, int level, BlockNumber blk, OffsetNumber coff)
-{
- Buffer buffer;
- Page page;
- GISTPageOpaque opaque;
- IndexTuple which;
- ItemId iid;
- OffsetNumber i,
- maxoff;
- BlockNumber cblk;
- char *pred;
-
- pred = (char *) palloc(sizeof(char) * level + 1);
- MemSet(pred, '\t', level);
- pred[level] = '\0';
-
- buffer = ReadBuffer(r, blk);
- page = (Page) BufferGetPage(buffer);
- opaque = (GISTPageOpaque) PageGetSpecialPointer(page);
-
- maxoff = PageGetMaxOffsetNumber(page);
-
- elog(DEBUG4, "%sPage: %d %s blk: %d maxoff: %d free: %d", pred,
- coff, (opaque->flags & F_LEAF) ? "LEAF" : "INTE", (int) blk,
- (int) maxoff, PageGetFreeSpace(page));
-
- for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
- {
- iid = PageGetItemId(page, i);
- which = (IndexTuple) PageGetItem(page, iid);
- cblk = ItemPointerGetBlockNumber(&(which->t_tid));
-#ifdef PRINTTUPLE
- elog(DEBUG4, "%s Tuple. blk: %d size: %d", pred, (int) cblk,
- IndexTupleSize(which));
-#endif
-
- if (!(opaque->flags & F_LEAF))
- gist_dumptree(r, level + 1, cblk, i);
- }
- ReleaseBuffer(buffer);
- pfree(pred);
-}
-#endif /* defined GISTDEBUG */
-
projects / postgresql.git / commitdiff