database/postgres
database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-05-06 19:59:18 +03:00
Code

Fix 'all at one page bug' in picksplit method of R-tree emulation. Add defense

Browse Source 
from buggy user-defined picksplit to GiST.
This commit is contained in:
Teodor Sigaev 2009-04-06 14:47:02 +00:00
parent 9462a22ab6
commit 4bc176deae
2 changed files with 175 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

123
src/backend/access/gist/gistproc.c
View File

@ -10,7 +10,7 @@
* Portions Copyright (c) 1994, Regents of the University of California * Portions Copyright (c) 1994, Regents of the University of California
* *
* IDENTIFICATION * IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/access/gist/gistproc.c,v 1.9.2.1 2007/09/07 17:04:46 teodor Exp $ * $PostgreSQL: pgsql/src/backend/access/gist/gistproc.c,v 1.9.2.2 2009/04/06 14:47:02 teodor Exp $
* *
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
@ -267,6 +267,69 @@ chooseLR(GIST_SPLITVEC *v,
v->spl_ldatum_exists = v->spl_rdatum_exists = false; v->spl_ldatum_exists = v->spl_rdatum_exists = false;
} }
/*
* Trivial split: half of entries will be placed on one page
* and another half - to another
*/
static void
fallbackSplit(GistEntryVector *entryvec, GIST_SPLITVEC *v)
{
OffsetNumber i,
maxoff;
BOX *unionL = NULL,
*unionR = NULL;
int nbytes;
maxoff = entryvec->n - 1;
nbytes = (maxoff + 2) * sizeof(OffsetNumber);
v->spl_left = (OffsetNumber *) palloc(nbytes);
v->spl_right = (OffsetNumber *) palloc(nbytes);
v->spl_nleft = v->spl_nright = 0;
for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
{
BOX * cur = DatumGetBoxP(entryvec->vector[i].key);
if (i <= (maxoff - FirstOffsetNumber + 1) / 2)
{
v->spl_left[v->spl_nleft] = i;
if (unionL == NULL)
{
unionL = (BOX *) palloc(sizeof(BOX));
*unionL = *cur;
}
else
adjustBox(unionL, cur);
v->spl_nleft++;
}
else
{
v->spl_right[v->spl_nright] = i;
if (unionR == NULL)
{
unionR = (BOX *) palloc(sizeof(BOX));
*unionR = *cur;
}
else
adjustBox(unionR, cur);
v->spl_nright++;
}
}
if (v->spl_ldatum_exists)
adjustBox(unionL, DatumGetBoxP(v->spl_ldatum));
v->spl_ldatum = BoxPGetDatum(unionL);
if (v->spl_rdatum_exists)
adjustBox(unionR, DatumGetBoxP(v->spl_rdatum));
v->spl_rdatum = BoxPGetDatum(unionR);
v->spl_ldatum_exists = v->spl_rdatum_exists = false;
}
/* /*
* The GiST PickSplit method * The GiST PickSplit method
* *
@ -319,52 +382,22 @@ gist_box_picksplit(PG_FUNCTION_ARGS)
adjustBox(&pageunion, cur); adjustBox(&pageunion, cur);
} }
if (allisequal)
{
/*
* All entries are the same
*/
fallbackSplit(entryvec, v);
PG_RETURN_POINTER(v);
}
nbytes = (maxoff + 2) * sizeof(OffsetNumber); nbytes = (maxoff + 2) * sizeof(OffsetNumber);
listL = (OffsetNumber *) palloc(nbytes); listL = (OffsetNumber *) palloc(nbytes);
listR = (OffsetNumber *) palloc(nbytes); listR = (OffsetNumber *) palloc(nbytes);
unionL = (BOX *) palloc(sizeof(BOX));
unionR = (BOX *) palloc(sizeof(BOX));
if (allisequal)
{
cur = DatumGetBoxP(entryvec->vector[OffsetNumberNext(FirstOffsetNumber)].key);
if (memcmp((void *) cur, (void *) &pageunion, sizeof(BOX)) == 0)
{
v->spl_left = listL;
v->spl_right = listR;
v->spl_nleft = v->spl_nright = 0;
memcpy((void *) unionL, (void *) &pageunion, sizeof(BOX));
memcpy((void *) unionR, (void *) &pageunion, sizeof(BOX));
for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
{
if (i <= (maxoff - FirstOffsetNumber + 1) / 2)
{
v->spl_left[v->spl_nleft] = i;
v->spl_nleft++;
}
else
{
v->spl_right[v->spl_nright] = i;
v->spl_nright++;
}
}
if (v->spl_ldatum_exists)
adjustBox(unionL, DatumGetBoxP(v->spl_ldatum));
v->spl_ldatum = BoxPGetDatum(unionL);
if (v->spl_rdatum_exists)
adjustBox(unionR, DatumGetBoxP(v->spl_rdatum));
v->spl_rdatum = BoxPGetDatum(unionR);
v->spl_ldatum_exists = v->spl_rdatum_exists = false;
PG_RETURN_POINTER(v);
}
}
listB = (OffsetNumber *) palloc(nbytes); listB = (OffsetNumber *) palloc(nbytes);
listT = (OffsetNumber *) palloc(nbytes); listT = (OffsetNumber *) palloc(nbytes);
unionL = (BOX *) palloc(sizeof(BOX));
unionR = (BOX *) palloc(sizeof(BOX));
unionB = (BOX *) palloc(sizeof(BOX)); unionB = (BOX *) palloc(sizeof(BOX));
unionT = (BOX *) palloc(sizeof(BOX)); unionT = (BOX *) palloc(sizeof(BOX));
@ -445,6 +478,12 @@ gist_box_picksplit(PG_FUNCTION_ARGS)
else else
ADDLIST(listT, unionT, posT, i); ADDLIST(listT, unionT, posT, i);
} }
if (IS_BADRATIO(posR, posL) && IS_BADRATIO(posT, posB))
{
fallbackSplit(entryvec, v);
PG_RETURN_POINTER(v);
}
} }
/* which split more optimal? */ /* which split more optimal? */

86
src/backend/access/gist/gistsplit.c
View File

@ -8,7 +8,7 @@
* Portions Copyright (c) 1994, Regents of the University of California * Portions Copyright (c) 1994, Regents of the University of California
* *
* IDENTIFICATION * IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/access/gist/gistsplit.c,v 1.3 2006/10/04 00:29:48 momjian Exp $ * $PostgreSQL: pgsql/src/backend/access/gist/gistsplit.c,v 1.3.2.1 2009/04/06 14:47:02 teodor Exp $
* *
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
@ -279,6 +279,63 @@ supportSecondarySplit(Relation r, GISTSTATE *giststate, int attno, GIST_SPLITVEC
sv->spl_ldatum_exists = sv->spl_rdatum_exists = false; sv->spl_ldatum_exists = sv->spl_rdatum_exists = false;
} }
/*
* Trivial picksplit implementaion. Function called only
* if user-defined picksplit puts all keys to the one page.
* That is a bug of user-defined picksplit but we'd like
* to "fix" that.
*/
static void
genericPickSplit(GISTSTATE *giststate, GistEntryVector *entryvec, GIST_SPLITVEC *v, int attno)
{
OffsetNumber i,
maxoff;
int nbytes;
GistEntryVector *evec;
maxoff = entryvec->n - 1;
nbytes = (maxoff + 2) * sizeof(OffsetNumber);
v->spl_left = (OffsetNumber *) palloc(nbytes);
v->spl_right = (OffsetNumber *) palloc(nbytes);
v->spl_nleft = v->spl_nright = 0;
for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
{
if (i <= (maxoff - FirstOffsetNumber + 1) / 2)
{
v->spl_left[v->spl_nleft] = i;
v->spl_nleft++;
}
else
{
v->spl_right[v->spl_nright] = i;
v->spl_nright++;
}
}
/*
* Form unions of each page
*/
evec = palloc( sizeof(GISTENTRY) * entryvec->n + GEVHDRSZ );
evec->n = v->spl_nleft;
memcpy(evec->vector, entryvec->vector + FirstOffsetNumber,
sizeof(GISTENTRY) * evec->n);
v->spl_ldatum = FunctionCall2(&giststate->unionFn[attno],
PointerGetDatum(evec),
PointerGetDatum(&nbytes));
evec->n = v->spl_nright;
memcpy(evec->vector, entryvec->vector + FirstOffsetNumber + v->spl_nleft,
sizeof(GISTENTRY) * evec->n);
v->spl_rdatum = FunctionCall2(&giststate->unionFn[attno],
PointerGetDatum(evec),
PointerGetDatum(&nbytes));
}
/* /*
* Calls user picksplit method for attno columns to split vector to * Calls user picksplit method for attno columns to split vector to
* two vectors. May use attno+n columns data to * two vectors. May use attno+n columns data to
@ -295,7 +352,7 @@ gistUserPicksplit(Relation r, GistEntryVector *entryvec, int attno, GistSplitVec
/* /*
* now let the user-defined picksplit function set up the split vector; in * now let the user-defined picksplit function set up the split vector; in
* entryvec have no null value!! * entryvec there is no null value!!
*/ */
sv->spl_ldatum_exists = (v->spl_lisnull[attno]) ? false : true; sv->spl_ldatum_exists = (v->spl_lisnull[attno]) ? false : true;
@ -307,6 +364,30 @@ gistUserPicksplit(Relation r, GistEntryVector *entryvec, int attno, GistSplitVec
PointerGetDatum(entryvec), PointerGetDatum(entryvec),
PointerGetDatum(sv)); PointerGetDatum(sv));
if ( sv->spl_nleft == 0 || sv->spl_nright == 0 )
{
ereport(DEBUG1,
(errcode(ERRCODE_INTERNAL_ERROR),
errmsg("Picksplit method for %d column of index \"%s\" failed",
attno+1, RelationGetRelationName(r)),
errhint("Index is not optimal, to optimize it contact developer or try to use the column as a second one in create index command")));
/*
* Reinit GIST_SPLITVEC. Although that fields are not used
* by genericPickSplit(), let us set up it for further processing
*/
sv->spl_ldatum_exists = (v->spl_lisnull[attno]) ? false : true;
sv->spl_rdatum_exists = (v->spl_risnull[attno]) ? false : true;
sv->spl_ldatum = v->spl_lattr[attno];
sv->spl_rdatum = v->spl_rattr[attno];
genericPickSplit(giststate, entryvec, sv, attno);
if (sv->spl_ldatum_exists || sv->spl_rdatum_exists)
supportSecondarySplit(r, giststate, attno, sv, v->spl_lattr[attno], v->spl_rattr[attno]);
}
else
{
/* compatibility with old code */ /* compatibility with old code */
if (sv->spl_left[sv->spl_nleft - 1] == InvalidOffsetNumber) if (sv->spl_left[sv->spl_nleft - 1] == InvalidOffsetNumber)
sv->spl_left[sv->spl_nleft - 1] = (OffsetNumber) (entryvec->n - 1); sv->spl_left[sv->spl_nleft - 1] = (OffsetNumber) (entryvec->n - 1);
@ -320,6 +401,7 @@ gistUserPicksplit(Relation r, GistEntryVector *entryvec, int attno, GistSplitVec
supportSecondarySplit(r, giststate, attno, sv, v->spl_lattr[attno], v->spl_rattr[attno]); supportSecondarySplit(r, giststate, attno, sv, v->spl_lattr[attno], v->spl_rattr[attno]);
} }
}
v->spl_lattr[attno] = sv->spl_ldatum; v->spl_lattr[attno] = sv->spl_ldatum;
v->spl_rattr[attno] = sv->spl_rdatum; v->spl_rattr[attno] = sv->spl_rdatum;