--- /dev/null
+/*-------------------------------------------------------------------------
+ *
+ * gistsplit.c
+ * Split page algorithm
+ *
+ *
+ * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * $PostgreSQL: pgsql/src/backend/access/gist/gistsplit.c,v 1.1 2006/06/28 12:08:35 teodor Exp $
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/genam.h"
+#include "access/gist_private.h"
+#include "access/gistscan.h"
+#include "access/heapam.h"
+#include "catalog/index.h"
+#include "commands/vacuum.h"
+#include "miscadmin.h"
+#include "utils/memutils.h"
+
+typedef struct {
+ Datum *attr;
+ int len;
+ OffsetNumber *entries;
+ bool *isnull;
+ bool *equiv;
+} GistSplitUnion;
+
+
+/*
+ * Forms unions of subkeys after page split, but
+ * uses only tuples aren't in groups of equalent tuples
+ */
+static void
+gistunionsubkeyvec(GISTSTATE *giststate, IndexTuple *itvec,
+ GistSplitUnion *gsvp, int startkey) {
+ IndexTuple *cleanedItVec;
+ int i, cleanedLen=0;
+
+ cleanedItVec = (IndexTuple*)palloc(sizeof(IndexTuple) * gsvp->len);
+
+ for(i=0;i<gsvp->len;i++) {
+ if ( gsvp->equiv && gsvp->equiv[gsvp->entries[i]])
+ continue;
+
+ cleanedItVec[cleanedLen++] = itvec[gsvp->entries[i] - 1];
+ }
+
+ gistMakeUnionItVec(giststate, cleanedItVec, cleanedLen, startkey,
+ gsvp->attr, gsvp->isnull);
+
+ pfree( cleanedItVec );
+}
+
+/*
+ * unions subkeys for after user picksplit over attno-1 column
+ */
+static void
+gistunionsubkey(GISTSTATE *giststate, IndexTuple *itvec, GistSplitVector *spl, int attno)
+{
+ GistSplitUnion gsvp;
+
+ gsvp.equiv = spl->spl_equiv;
+
+ gsvp.attr = spl->spl_lattr;
+ gsvp.len = spl->splitVector.spl_nleft;
+ gsvp.entries = spl->splitVector.spl_left;
+ gsvp.isnull = spl->spl_lisnull;
+
+ gistunionsubkeyvec(giststate, itvec, &gsvp, attno);
+
+ gsvp.attr = spl->spl_rattr;
+ gsvp.len = spl->splitVector.spl_nright;
+ gsvp.entries = spl->splitVector.spl_right;
+ gsvp.isnull = spl->spl_risnull;
+
+ gistunionsubkeyvec(giststate, itvec, &gsvp, attno);
+}
+
+/*
+ * find group in vector with equivalent value
+ */
+static int
+gistfindgroup(Relation r, GISTSTATE *giststate, GISTENTRY *valvec, GistSplitVector *spl, int attno)
+{
+ int i;
+ GISTENTRY entry;
+ int len=0;
+
+ /*
+ * attno key is always not null (see gistSplitByKey), so we may not check for
+ * nulls
+ */
+ gistentryinit(entry, spl->splitVector.spl_rdatum, r, NULL, (OffsetNumber) 0, FALSE);
+ for (i = 0; i < spl->splitVector.spl_nleft; i++) {
+ float penalty = gistpenalty(giststate, attno, &entry, false,
+ &valvec[spl->splitVector.spl_left[i]], false);
+ if ( penalty == 0.0 ) {
+ spl->spl_equiv[spl->splitVector.spl_left[i]] = true;
+ len++;
+ }
+ }
+
+ gistentryinit(entry, spl->splitVector.spl_ldatum, r, NULL, (OffsetNumber) 0, FALSE);
+ for (i = 0; i < spl->splitVector.spl_nright; i++) {
+ float penalty = gistpenalty(giststate, attno, &entry, false,
+ &valvec[spl->splitVector.spl_right[i]], false);
+ if ( penalty == 0.0 ) {
+ spl->spl_equiv[spl->splitVector.spl_right[i]] = true;
+ len++;
+ }
+ }
+
+ return len;
+}
+
+static void
+cleanupOffsets( OffsetNumber *a, int *len, bool *equiv, int *LenEquiv ) {
+ int curlen,i;
+ OffsetNumber *curwpos;
+
+ curlen = *len;
+ curwpos = a;
+ for (i = 0; i < *len; i++) {
+ if ( equiv[ a[i] ] == FALSE ) {
+ *curwpos = a[i];
+ curwpos++;
+ } else {
+ /* corner case: we shouldn't make void array */
+ if ( curlen==1 ) {
+ equiv[ a[i] ] = FALSE; /* mark item as non-equivalent */
+ i--; /* redo the same */
+ *LenEquiv -= 1;
+ continue;
+ } else
+ curlen--;
+ }
+ }
+
+ *len = curlen;
+}
+
+static void
+placeOne( Relation r, GISTSTATE *giststate, GistSplitVector *v, IndexTuple itup, OffsetNumber off, int attno ) {
+ GISTENTRY identry[INDEX_MAX_KEYS];
+ bool isnull[INDEX_MAX_KEYS];
+ bool toLeft = true;
+
+ gistDeCompressAtt(giststate, r, itup, NULL, (OffsetNumber) 0, identry, isnull);
+
+ for(;attno<giststate->tupdesc->natts;attno++) {
+ float lpenalty, rpenalty;
+ GISTENTRY entry;
+
+ gistentryinit(entry, v->spl_lattr[attno], r, NULL, 0, FALSE);
+ lpenalty = gistpenalty(giststate, attno, &entry, v->spl_lisnull[attno], identry+attno, isnull[ attno ]);
+ gistentryinit(entry, v->spl_rattr[attno], r, NULL, 0, FALSE);
+ rpenalty = gistpenalty(giststate, attno, &entry, v->spl_risnull[attno], identry+attno, isnull[ attno ]);
+
+ if ( lpenalty != rpenalty ) {
+ if ( lpenalty > rpenalty )
+ toLeft = false;
+ break;
+ }
+ }
+
+ if ( toLeft )
+ v->splitVector.spl_left[ v->splitVector.spl_nleft++ ] = off;
+ else
+ v->splitVector.spl_right[ v->splitVector.spl_nright++ ] = off;
+}
+
+#define SWAPVAR( s, d, t ) \
+do { \
+ (t) = (s); \
+ (s) = (d); \
+ (d) = (t); \
+} while(0)
+
+/*
+ * adjust left and right unions according to splits by previous
+ * split by firsts columns. This function is called only in case
+ * when pickSplit doesn't support subspplit.
+ */
+
+static void
+supportSecondarySplit( Relation r, GISTSTATE *giststate, int attno, GIST_SPLITVEC *sv, Datum oldL, Datum oldR ) {
+ bool leaveOnLeft = true, tmpBool;
+ GISTENTRY entryL, entryR, entrySL, entrySR;
+
+ gistentryinit(entryL, oldL, r, NULL, 0, FALSE);
+ gistentryinit(entryR, oldR, r, NULL, 0, FALSE);
+ gistentryinit(entrySL, sv->spl_ldatum , r, NULL, 0, FALSE);
+ gistentryinit(entrySR, sv->spl_rdatum , r, NULL, 0, FALSE);
+
+ if ( sv->spl_ldatum_exists && sv->spl_rdatum_exists ) {
+ float penalty1, penalty2;
+
+ penalty1 = gistpenalty(giststate, attno, &entryL, false, &entrySL, false) +
+ gistpenalty(giststate, attno, &entryR, false, &entrySR, false);
+ penalty2 = gistpenalty(giststate, attno, &entryL, false, &entrySR, false) +
+ gistpenalty(giststate, attno, &entryR, false, &entrySL, false);
+
+ if ( penalty1 > penalty2 )
+ leaveOnLeft = false;
+
+ } else {
+ GISTENTRY *entry1 = (sv->spl_ldatum_exists) ? &entryL : &entryR;
+ float penalty1, penalty2;
+
+ /*
+ * there is only one previously defined union,
+ * so we just choose swap or not by lowest penalty
+ */
+
+ penalty1 = gistpenalty(giststate, attno, entry1, false, &entrySL, false);
+ penalty2 = gistpenalty(giststate, attno, entry1, false, &entrySR, false);
+
+ if ( penalty1 < penalty2 )
+ leaveOnLeft = ( sv->spl_ldatum_exists ) ? true : false;
+ else
+ leaveOnLeft = ( sv->spl_rdatum_exists ) ? true : false;
+ }
+
+ if ( leaveOnLeft == false ) {
+ /*
+ * swap left and right
+ */
+ OffsetNumber *off, noff;
+ Datum datum;
+
+ SWAPVAR( sv->spl_left, sv->spl_right, off );
+ SWAPVAR( sv->spl_nleft, sv->spl_nright, noff );
+ SWAPVAR( sv->spl_ldatum, sv->spl_rdatum, datum );
+ gistentryinit(entrySL, sv->spl_ldatum , r, NULL, 0, FALSE);
+ gistentryinit(entrySR, sv->spl_rdatum , r, NULL, 0, FALSE);
+ }
+
+ if ( sv->spl_ldatum_exists )
+ gistMakeUnionKey(giststate, attno, &entryL, false, &entrySL, false,
+ &sv->spl_ldatum, &tmpBool);
+
+ if ( sv->spl_rdatum_exists )
+ gistMakeUnionKey(giststate, attno, &entryR, false, &entrySR, false,
+ &sv->spl_rdatum, &tmpBool);
+
+ sv->spl_ldatum_exists = sv->spl_rdatum_exists = false;
+}
+
+/*
+ * Calls user picksplit method for attno columns to split vector to
+ * two vectors. May use attno+n columns data to
+ * get better split.
+ * Returns TRUE and v->spl_equiv = NULL if left and right unions of attno columns are the same,
+ * so caller may find better split
+ * Returns TRUE and v->spl_equiv != NULL if there is tuples which may be freely moved
+ */
+static bool
+gistUserPicksplit(Relation r, GistEntryVector *entryvec, int attno, GistSplitVector *v,
+ IndexTuple *itup, int len, GISTSTATE *giststate)
+{
+ GIST_SPLITVEC *sv = &v->splitVector;
+ /*
+ * now let the user-defined picksplit function set up the split vector; in
+ * entryvec have no null value!!
+ */
+
+ 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];
+
+ FunctionCall2(&giststate->picksplitFn[attno],
+ PointerGetDatum(entryvec),
+ PointerGetDatum(sv));
+
+ /* compatibility with old code */
+ if (sv->spl_left[sv->spl_nleft - 1] == InvalidOffsetNumber)
+ sv->spl_left[sv->spl_nleft - 1] = (OffsetNumber) (entryvec->n - 1);
+ if (sv->spl_right[sv->spl_nright - 1] == InvalidOffsetNumber)
+ sv->spl_right[sv->spl_nright - 1] = (OffsetNumber) (entryvec->n - 1);
+
+ if( sv->spl_ldatum_exists || sv->spl_rdatum_exists ) {
+ elog(LOG,"PickSplit method of %d columns of index '%s' doesn't support secondary split",
+ attno + 1, RelationGetRelationName(r) );
+
+ supportSecondarySplit( r, giststate, attno, sv, v->spl_lattr[attno], v->spl_rattr[attno] );
+ }
+
+ v->spl_lattr[attno] = sv->spl_ldatum;
+ v->spl_rattr[attno] = sv->spl_rdatum;
+ v->spl_lisnull[attno] = false;
+ v->spl_risnull[attno] = false;
+
+ /*
+ * if index is multikey, then we must to try get smaller bounding box for
+ * subkey(s)
+ */
+ v->spl_equiv = NULL;
+
+ if (giststate->tupdesc->natts > 1 && attno+1 != giststate->tupdesc->natts)
+ {
+ if ( gistKeyIsEQ(giststate, attno, sv->spl_ldatum, sv->spl_rdatum) ) {
+ /*
+ * Left and right key's unions are equial, so
+ * we can get better split by following columns. Note,
+ * unions for attno columns are already done.
+ */
+
+ return true;
+ } else {
+ int LenEquiv;
+
+ v->spl_equiv = (bool *) palloc0(sizeof(bool) * (entryvec->n+1));
+
+ LenEquiv = gistfindgroup(r, giststate, entryvec->vector, v, attno);
+
+ /*
+ * if possible, we should distribute equivalent tuples
+ */
+ if (LenEquiv == 0 ) {
+ gistunionsubkey(giststate, itup, v, attno + 1);
+ } else {
+ cleanupOffsets( sv->spl_left, &sv->spl_nleft, v->spl_equiv, &LenEquiv );
+ cleanupOffsets( sv->spl_right, &sv->spl_nright, v->spl_equiv, &LenEquiv );
+
+ gistunionsubkey(giststate, itup, v, attno + 1);
+ if (LenEquiv == 1 ) {
+ /*
+ * In case with one tuple we just choose left-right
+ * by penalty. It's simplify user-defined pickSplit
+ */
+ OffsetNumber toMove = InvalidOffsetNumber;
+
+ for(toMove=FirstOffsetNumber;toMove<entryvec->n;toMove++)
+ if ( v->spl_equiv[ toMove ] )
+ break;
+ Assert( toMove < entryvec->n );
+
+ placeOne( r, giststate, v, itup[ toMove-1 ], toMove, attno+1 );
+ /* redo gistunionsubkey(): it will not degradate performance,
+ * because it's very rarely */
+ v->spl_equiv = NULL;
+ gistunionsubkey(giststate, itup, v, attno + 1);
+
+ return false;
+ } else if ( LenEquiv > 1 )
+ return true;
+ }
+ }
+ }
+
+ return false;
+}
+
+/*
+ * simple split page
+ */
+static void
+gistSplitHalf(GIST_SPLITVEC *v, int len) {
+ int i;
+
+ v->spl_nright = v->spl_nleft = 0;
+ v->spl_left = (OffsetNumber *) palloc(len * sizeof(OffsetNumber));
+ v->spl_right= (OffsetNumber *) palloc(len * sizeof(OffsetNumber));
+ for(i = 1; i <= len; i++)
+ if ( i<len/2 )
+ v->spl_right[ v->spl_nright++ ] = i;
+ else
+ v->spl_left[ v->spl_nleft++ ] = i;
+}
+
+/*
+ * if it was invalid tuple then we need special processing.
+ * We move all invalid tuples on right page.
+ *
+ * if there is no place on left page, gistSplit will be called one more
+ * time for left page.
+ *
+ * Normally, we never exec this code, but after crash replay it's possible
+ * to get 'invalid' tuples (probability is low enough)
+ */
+static void
+gistSplitByInvalid(GISTSTATE *giststate, GistSplitVector *v, IndexTuple *itup, int len) {
+ int i;
+ static OffsetNumber offInvTuples[ MaxOffsetNumber ];
+ int nOffInvTuples = 0;
+
+ for (i = 1; i <= len; i++)
+ if ( GistTupleIsInvalid(itup[i - 1]) )
+ offInvTuples[ nOffInvTuples++ ] = i;
+
+ if ( nOffInvTuples == len ) {
+ /* corner case, all tuples are invalid */
+ v->spl_rightvalid= v->spl_leftvalid = false;
+ gistSplitHalf( &v->splitVector, len );
+ } else {
+ GistSplitUnion gsvp;
+
+ v->splitVector.spl_right = offInvTuples;
+ v->splitVector.spl_nright = nOffInvTuples;
+ v->spl_rightvalid = false;
+
+ v->splitVector.spl_left = (OffsetNumber *) palloc(len * sizeof(OffsetNumber));
+ v->splitVector.spl_nleft = 0;
+ for(i = 1; i <= len; i++)
+ if ( !GistTupleIsInvalid(itup[i - 1]) )
+ v->splitVector.spl_left[ v->splitVector.spl_nleft++ ] = i;
+ v->spl_leftvalid = true;
+
+ gsvp.equiv = NULL;
+ gsvp.attr = v->spl_lattr;
+ gsvp.len = v->splitVector.spl_nleft;
+ gsvp.entries = v->splitVector.spl_left;
+ gsvp.isnull = v->spl_lisnull;
+
+ gistunionsubkeyvec(giststate, itup, &gsvp, 0);
+ }
+}
+
+/*
+ * trys to split page by attno key, in a case of null
+ * values move its to separate page.
+ */
+void
+gistSplitByKey(Relation r, Page page, IndexTuple *itup, int len, GISTSTATE *giststate,
+ GistSplitVector *v, GistEntryVector *entryvec, int attno) {
+ int i;
+ static OffsetNumber offNullTuples[ MaxOffsetNumber ];
+ int nOffNullTuples = 0;
+
+ for (i = 1; i <= len; i++) {
+ Datum datum;
+ bool IsNull;
+
+ if (!GistPageIsLeaf(page) && GistTupleIsInvalid(itup[i - 1])) {
+ gistSplitByInvalid(giststate, v, itup, len);
+ return;
+ }
+
+ datum = index_getattr(itup[i - 1], attno+1, giststate->tupdesc, &IsNull);
+ gistdentryinit(giststate, attno, &(entryvec->vector[i]),
+ datum, r, page, i,
+ FALSE, IsNull);
+ if ( IsNull )
+ offNullTuples[ nOffNullTuples++ ] = i;
+ }
+
+ v->spl_leftvalid = v->spl_rightvalid = true;
+
+ if ( nOffNullTuples == len ) {
+ /*
+ * Corner case: All keys in attno column are null, we should try to
+ * split by keys in next column. It all keys in all columns
+ * are NULL just split page half by half
+ */
+ v->spl_risnull[attno] = v->spl_lisnull[attno] = TRUE;
+
+ if ( attno+1 == r->rd_att->natts )
+ gistSplitHalf( &v->splitVector, len );
+ else
+ gistSplitByKey(r, page, itup, len, giststate, v, entryvec, attno+1);
+ } else if ( nOffNullTuples > 0 ) {
+ int j=0;
+
+ /*
+ * We don't want to mix NULLs and not-NULLs keys
+ * on one page, so move nulls to right page
+ */
+ v->splitVector.spl_right = offNullTuples;
+ v->splitVector.spl_nright = nOffNullTuples;
+ v->spl_risnull[attno] = TRUE;
+
+ v->splitVector.spl_left = (OffsetNumber *) palloc(len * sizeof(OffsetNumber));
+ v->splitVector.spl_nleft = 0;
+ for(i = 1; i <= len; i++)
+ if ( j<v->splitVector.spl_nright && offNullTuples[j] == i )
+ j++;
+ else
+ v->splitVector.spl_left[ v->splitVector.spl_nleft++ ] = i;
+
+ v->spl_equiv = NULL;
+ gistunionsubkey(giststate, itup, v, attno);
+ } else {
+ /*
+ * all keys are not-null
+ */
+ entryvec->n = len+1;
+
+ if ( gistUserPicksplit(r, entryvec, attno, v, itup, len, giststate) && attno+1 != r->rd_att->natts ) {
+ /*
+ * Splitting on attno column is not optimized: there is a tuples which can be freely
+ * left or right page, we will try to split page by
+ * following columns
+ */
+ if ( v->spl_equiv == NULL ) {
+ /* simple case: left and right keys for attno column are equial */
+ gistSplitByKey(r, page, itup, len, giststate, v, entryvec, attno+1);
+ } else {
+ /* we should clean up vector from already distributed tuples */
+ IndexTuple *newitup = (IndexTuple*)palloc((len + 1) * sizeof(IndexTuple));
+ OffsetNumber *map = (OffsetNumber*)palloc((len + 1) * sizeof(IndexTuple));
+ int newlen = 0;
+ GIST_SPLITVEC backupSplit = v->splitVector;
+
+ for(i=0; i<len; i++)
+ if ( v->spl_equiv[i+1] ) {
+ map[ newlen ] = i+1;
+ newitup[ newlen++ ] = itup[i];
+ }
+
+ Assert( newlen>0 );
+
+ backupSplit.spl_left = (OffsetNumber*)palloc(sizeof(OffsetNumber)*len);
+ memcpy( backupSplit.spl_left, v->splitVector.spl_left, sizeof(OffsetNumber)*v->splitVector.spl_nleft);
+ backupSplit.spl_right = (OffsetNumber*)palloc(sizeof(OffsetNumber)*len);
+ memcpy( backupSplit.spl_right, v->splitVector.spl_right, sizeof(OffsetNumber)*v->splitVector.spl_nright);
+
+ gistSplitByKey(r, page, newitup, newlen, giststate, v, entryvec, attno+1);
+
+ /* merge result of subsplit */
+ for(i=0;i<v->splitVector.spl_nleft;i++)
+ backupSplit.spl_left[ backupSplit.spl_nleft++ ] = map[ v->splitVector.spl_left[i]-1 ];
+ for(i=0;i<v->splitVector.spl_nright;i++)
+ backupSplit.spl_right[ backupSplit.spl_nright++ ] = map[ v->splitVector.spl_right[i]-1 ];
+
+ v->splitVector = backupSplit;
+ /* reunion left and right datums */
+ gistunionsubkey(giststate, itup, v, attno);
+ }
+ }
+ }
+}
projects / postgresql / commitdiff