Changes

* new split algorithm (as proposed in http://archives.postgresql.org/pgsql-hackers/2006-06/msg00254.php)
  * possible call pickSplit() for second and below columns
  * add spl_(l|r)datum_exists to GIST_SPLITVEC -
    pickSplit should check its values to use already defined
    spl_(l|r)datum for splitting. pickSplit should set
    spl_(l|r)datum_exists to 'false' (if they was 'true') to
    signal to caller about using spl_(l|r)datum.
  * support for old pickSplit(): not very optimal
    but correct split
* remove 'bytes' field from GISTENTRY: in any case size of
  value is defined by it's type.
* split GIST_SPLITVEC to two structures: one for using in picksplit
  and second - for internal use.
* some code refactoring
* support of subsplit to rtree opclasses

TODO: add support of subsplit to contrib modules

contrib/btree_gist/btree_inet.c

@@ -100,7 +100,7 @@ gbt_inet_compress(PG_FUNCTION_ARGS)
 		r->upper = r->lower;
 		gistentryinit(*retval, PointerGetDatum(r),
 					  entry->rel, entry->page,
-					  entry->offset, sizeof(inetKEY), FALSE);
+					  entry->offset, FALSE);
 	}
 	else
 		retval = entry;

contrib/btree_gist/btree_interval.c

@@ -129,7 +129,7 @@ gbt_intv_compress(PG_FUNCTION_ARGS)
 		}
 		gistentryinit(*retval, PointerGetDatum(r),
 					  entry->rel, entry->page,
-					  entry->offset, 2 * INTERVALSIZE, FALSE);
+					  entry->offset, FALSE);
 	}
 
 	PG_RETURN_POINTER(retval);
@@ -153,7 +153,7 @@ gbt_intv_decompress(PG_FUNCTION_ARGS)
 
 		gistentryinit(*retval, PointerGetDatum(r),
 					  entry->rel, entry->page,
-					  entry->offset, sizeof(intvKEY), FALSE);
+					  entry->offset, FALSE);
 	}
 	PG_RETURN_POINTER(retval);
 }

contrib/btree_gist/btree_text.c

@@ -115,7 +115,7 @@ gbt_bpchar_compress(PG_FUNCTION_ARGS)
 
 		gistentryinit(trim, d,
 					  entry->rel, entry->page,
-					  entry->offset, VARSIZE(DatumGetPointer(d)), TRUE);
+					  entry->offset, TRUE);
 		retval = gbt_var_compress(&trim, &tinfo);
 	}
 	else

contrib/btree_gist/btree_time.c

@@ -137,7 +137,7 @@ gbt_timetz_compress(PG_FUNCTION_ARGS)
 		r->lower = r->upper = tmp;
 		gistentryinit(*retval, PointerGetDatum(r),
 					  entry->rel, entry->page,
-					  entry->offset, sizeof(timeKEY), FALSE);
+					  entry->offset, FALSE);
 	}
 	else
 		retval = entry;

contrib/btree_gist/btree_ts.c

@@ -159,7 +159,7 @@ gbt_tstz_compress(PG_FUNCTION_ARGS)
 		r->lower = r->upper = gmt;
 		gistentryinit(*retval, PointerGetDatum(r),
 					  entry->rel, entry->page,
-					  entry->offset, sizeof(tsKEY), FALSE);
+					  entry->offset, FALSE);
 	}
 	else
 		retval = entry;

contrib/btree_gist/btree_utils_num.c

@@ -46,7 +46,7 @@ gbt_num_compress(GISTENTRY *retval, GISTENTRY *entry, const gbtree_ninfo * tinfo
 		memcpy((void *) &r[tinfo->size], leaf, tinfo->size);
 		retval = palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(r), entry->rel, entry->page,
-					  entry->offset, (2 * tinfo->size), FALSE);
+					  entry->offset, FALSE);
 	}
 	else
 		retval = entry;

contrib/btree_gist/btree_utils_var.c

@@ -19,7 +19,7 @@ gbt_var_decompress(PG_FUNCTION_ARGS)
 
 		gistentryinit(*retval, PointerGetDatum(key),
 					  entry->rel, entry->page,
-					  entry->offset, VARSIZE(key), FALSE);
+					  entry->offset, FALSE);
 
 		PG_RETURN_POINTER(retval);
 	}
@@ -292,7 +292,7 @@ gbt_var_compress(GISTENTRY *entry, const gbtree_vinfo * tinfo)
 		retval = palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(r),
 					  entry->rel, entry->page,
-					  entry->offset, VARSIZE(r), TRUE);
+					  entry->offset, TRUE);
 	}
 	else
 		retval = entry;

contrib/cube/cube.c

 /******************************************************************************
-  $PostgreSQL: pgsql/contrib/cube/cube.c,v 1.25 2006/05/30 22:12:12 tgl Exp $
+  $PostgreSQL: pgsql/contrib/cube/cube.c,v 1.26 2006/06/28 11:59:59 teodor Exp $
 
   This file contains routines that can be bound to a Postgres backend and
   called by the backend in the process of processing queries.  The calling
@@ -300,8 +300,8 @@ g_cube_picksplit(GistEntryVector *entryvec,
 	double		size_l,
 				size_r;
 	int			nbytes;
-	OffsetNumber seed_1 = 0,
-				seed_2 = 0;
+	OffsetNumber seed_1 = 1,
+				seed_2 = 2;
 	OffsetNumber *left,
 			   *right;
 	OffsetNumber maxoff;

contrib/intarray/_int_gist.c

@@ -154,7 +154,7 @@ g_int_compress(PG_FUNCTION_ARGS)
 
 		retval = palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(r),
-				  entry->rel, entry->page, entry->offset, VARSIZE(r), FALSE);
+				  entry->rel, entry->page, entry->offset, FALSE);
 
 		PG_RETURN_POINTER(retval);
 	}
@@ -201,7 +201,7 @@ g_int_compress(PG_FUNCTION_ARGS)
 		r = resize_intArrayType(r, len);
 		retval = palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(r),
-				  entry->rel, entry->page, entry->offset, VARSIZE(r), FALSE);
+				  entry->rel, entry->page, entry->offset, FALSE);
 		PG_RETURN_POINTER(retval);
 	}
 	else
@@ -238,7 +238,7 @@ g_int_decompress(PG_FUNCTION_ARGS)
 		{
 			retval = palloc(sizeof(GISTENTRY));
 			gistentryinit(*retval, PointerGetDatum(in),
-				 entry->rel, entry->page, entry->offset, VARSIZE(in), FALSE);
+				 entry->rel, entry->page, entry->offset, FALSE);
 
 			PG_RETURN_POINTER(retval);
 		}
@@ -260,7 +260,7 @@ g_int_decompress(PG_FUNCTION_ARGS)
 		pfree(in);
 	retval = palloc(sizeof(GISTENTRY));
 	gistentryinit(*retval, PointerGetDatum(r),
-				  entry->rel, entry->page, entry->offset, VARSIZE(r), FALSE);
+				  entry->rel, entry->page, entry->offset, FALSE);
 
 	PG_RETURN_POINTER(retval);
 }

contrib/intarray/_intbig_gist.c

@@ -172,7 +172,7 @@ g_intbig_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(res),
 					  entry->rel, entry->page,
-					  entry->offset, res->len, FALSE);
+					  entry->offset, FALSE);
 
 		if (in != (ArrayType *) PG_DETOAST_DATUM(entry->key))
 			pfree(in);
@@ -198,7 +198,7 @@ g_intbig_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(res),
 					  entry->rel, entry->page,
-					  entry->offset, res->len, FALSE);
+					  entry->offset, FALSE);
 
 		PG_RETURN_POINTER(retval);
 	}

contrib/ltree/_ltree_gist.c

@@ -108,7 +108,7 @@ _ltree_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(key),
 					  entry->rel, entry->page,
-					  entry->offset, key->len, FALSE);
+					  entry->offset, FALSE);
 	}
 	else if (!LTG_ISALLTRUE(entry->key))
 	{
@@ -130,7 +130,7 @@ _ltree_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(key),
 					  entry->rel, entry->page,
-					  entry->offset, key->len, FALSE);
+					  entry->offset, FALSE);
 	}
 	PG_RETURN_POINTER(retval);
 }

contrib/ltree/ltree_gist.c

 /*
  * GiST support for ltree
  * Teodor Sigaev <teodor@stack.net>
- * $PostgreSQL: pgsql/contrib/ltree/ltree_gist.c,v 1.14 2006/03/11 04:38:29 momjian Exp $
+ * $PostgreSQL: pgsql/contrib/ltree/ltree_gist.c,v 1.15 2006/06/28 11:59:59 teodor Exp $
  */
 
 #include "ltree.h"
@@ -81,7 +81,7 @@ ltree_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(key),
 					  entry->rel, entry->page,
-					  entry->offset, key->len, FALSE);
+					  entry->offset, FALSE);
 	}
 	PG_RETURN_POINTER(retval);
 }
@@ -98,7 +98,7 @@ ltree_decompress(PG_FUNCTION_ARGS)
 
 		gistentryinit(*retval, PointerGetDatum(key),
 					  entry->rel, entry->page,
-					  entry->offset, key->len, FALSE);
+					  entry->offset, FALSE);
 		PG_RETURN_POINTER(retval);
 	}
 	PG_RETURN_POINTER(entry);

contrib/pg_trgm/trgm_gist.c

@@ -103,7 +103,7 @@ gtrgm_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(res),
 					  entry->rel, entry->page,
-					  entry->offset, res->len, FALSE);
+					  entry->offset, FALSE);
 	}
 	else if (ISSIGNKEY(DatumGetPointer(entry->key)) &&
 			 !ISALLTRUE(DatumGetPointer(entry->key)))
@@ -126,7 +126,7 @@ gtrgm_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(res),
 					  entry->rel, entry->page,
-					  entry->offset, res->len, FALSE);
+					  entry->offset, FALSE);
 	}
 	PG_RETURN_POINTER(retval);
 }

contrib/seg/seg.c

@@ -316,8 +316,8 @@ gseg_picksplit(GistEntryVector *entryvec,
 	float		size_l,
 				size_r;
 	int			nbytes;
-	OffsetNumber seed_1 = 0,
-				seed_2 = 0;
+	OffsetNumber seed_1 = 1,
+				seed_2 = 2;
 	OffsetNumber *left,
 			   *right;
 	OffsetNumber maxoff;

contrib/tsearch2/gistidx.c

-/* $PostgreSQL: pgsql/contrib/tsearch2/gistidx.c,v 1.13 2006/03/11 04:38:30 momjian Exp $ */
+/* $PostgreSQL: pgsql/contrib/tsearch2/gistidx.c,v 1.14 2006/06/28 12:00:06 teodor Exp $ */
 
 #include "postgres.h"
 
@@ -202,7 +202,7 @@ gtsvector_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(res),
 					  entry->rel, entry->page,
-					  entry->offset, res->len, FALSE);
+					  entry->offset, FALSE);
 	}
 	else if (ISSIGNKEY(DatumGetPointer(entry->key)) &&
 			 !ISALLTRUE(DatumGetPointer(entry->key)))
@@ -225,7 +225,7 @@ gtsvector_compress(PG_FUNCTION_ARGS)
 		retval = (GISTENTRY *) palloc(sizeof(GISTENTRY));
 		gistentryinit(*retval, PointerGetDatum(res),
 					  entry->rel, entry->page,
-					  entry->offset, res->len, FALSE);
+					  entry->offset, FALSE);
 	}
 	PG_RETURN_POINTER(retval);
 }
@@ -242,7 +242,7 @@ gtsvector_decompress(PG_FUNCTION_ARGS)
 
 		gistentryinit(*retval, PointerGetDatum(key),
 					  entry->rel, entry->page,
-					  entry->offset, key->len, FALSE);
+					  entry->offset, FALSE);
 
 		PG_RETURN_POINTER(retval);
 	}

contrib/tsearch2/query_gist.c

@@ -164,7 +164,7 @@ gtsq_compress(PG_FUNCTION_ARGS)
 
 		gistentryinit(*retval, PointerGetDatum(sign),
 					  entry->rel, entry->page,
-					  entry->offset, sizeof(TPQTGist), FALSE);
+					  entry->offset, FALSE);
 	}
 
 	PG_RETURN_POINTER(retval);

src/backend/access/gist/Makefile

@@ -4,7 +4,7 @@
 #    Makefile for access/gist
 #
 # IDENTIFICATION
-#    $PostgreSQL: pgsql/src/backend/access/gist/Makefile,v 1.15 2005/07/01 19:19:02 tgl Exp $
+#    $PostgreSQL: pgsql/src/backend/access/gist/Makefile,v 1.16 2006/06/28 12:00:13 teodor Exp $
 #
 #-------------------------------------------------------------------------
 
@@ -13,7 +13,7 @@ top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = gist.o gistutil.o gistxlog.o gistvacuum.o gistget.o gistscan.o \
-       gistproc.o
+       gistproc.o gistsplit.o
 
 all: SUBSYS.o
 

src/backend/access/gist/gist.c

@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/gist/gist.c,v 1.138 2006/05/29 12:50:06 teodor Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/gist/gist.c,v 1.139 2006/06/28 12:00:14 teodor Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -187,7 +187,7 @@ gistbuildCallback(Relation index,
 
 	/* form an index tuple and point it at the heap tuple */
 	itup = gistFormTuple(&buildstate->giststate, index,
-		values, NULL /* size is currently bogus */, isnull);
+		values, isnull, true /* size is currently bogus */);
 	itup->t_tid = htup->t_self;
 
 	/*
@@ -233,7 +233,7 @@ gistinsert(PG_FUNCTION_ARGS)
 	initGISTstate(&giststate, r);
 
 	itup = gistFormTuple(&giststate, r,
-		values, NULL /* size is currently bogus */, isnull);
+		values, isnull, true /* size is currently bogus */);
 	itup->t_tid = *ht_ctid;
 
 	gistdoinsert(r, itup, &giststate);
@@ -899,150 +899,6 @@ gistmakedeal(GISTInsertState *state, GISTSTATE *giststate)
 		gistxlogInsertCompletion(state->r->rd_node, &(state->key), 1);
 }
 
-/*
- * 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, GIST_SPLITVEC *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, len );
-	} else {
-		GistSplitVec    gsvp;
-				
-		v->spl_right = offInvTuples;
-		v->spl_nright = nOffInvTuples;
-		v->spl_rightvalid = false;
-
-		v->spl_left = (OffsetNumber *) palloc(len * sizeof(OffsetNumber));
-		v->spl_nleft = 0;
-		for(i = 1; i <= len; i++) 
-			if ( !GistTupleIsInvalid(itup[i - 1]) )
-				v->spl_left[ v->spl_nleft++ ] = i;
-		v->spl_leftvalid = true;
-		
-		gsvp.idgrp = NULL;
-		gsvp.attrsize = v->spl_lattrsize;
-		gsvp.attr = v->spl_lattr;
-		gsvp.len = v->spl_nleft;
-		gsvp.entries = v->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. 
- */
-static void
-gistSplitByKey(Relation r, Page page, IndexTuple *itup, int len, GISTSTATE *giststate, 
-		GIST_SPLITVEC *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,
-					   ATTSIZE(datum, giststate->tupdesc, attno+1, IsNull),
-					   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
-		 * 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, 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->spl_right = offNullTuples;
-		v->spl_nright = nOffNullTuples;
-		v->spl_risnull[attno] = TRUE;
-
-		v->spl_left = (OffsetNumber *) palloc(len * sizeof(OffsetNumber));
-		v->spl_nleft = 0;
-		for(i = 1; i <= len; i++) 
-			if ( j<v->spl_nright && offNullTuples[j] == i ) 
-				j++;
-			else
-				v->spl_left[ v->spl_nleft++ ] = i;
-
-		v->spl_idgrp = NULL;
-		gistunionsubkey(giststate, itup, v, 0);
-	} else {
-		/*
-		 * all keys are not-null
-		 */
-		if ( gistUserPicksplit(r, entryvec, attno, v, itup, len, giststate) && attno+1 != r->rd_att->natts )
-			/*
-			 * Splitting on attno column is not optimized: unions of left and right
-			 * page are the same, we will try to split page by 
-			 * following columns
-			 */
-			gistSplitByKey(r, page, itup, len, giststate, v, entryvec, attno+1);
-	}
-}
-
 /*
  * gistSplit -- split a page in the tree and fill struct
  * used for XLOG and real writes buffers. Function is recursive, ie
@@ -1057,7 +913,7 @@ gistSplit(Relation r,
 {
 	IndexTuple *lvectup,
 			   *rvectup;
-	GIST_SPLITVEC v;
+	GistSplitVector v;
 	GistEntryVector *entryvec;
 	int			i;
 	SplitedPageLayout	*res = NULL;
@@ -1066,6 +922,8 @@ gistSplit(Relation r,
 	entryvec = palloc(GEVHDRSZ + (len + 1) * sizeof(GISTENTRY));
 	entryvec->n = len + 1;
 
+	memset( v.spl_lisnull, TRUE, sizeof(bool) * giststate->tupdesc->natts ); 
+	memset( v.spl_risnull, TRUE, sizeof(bool) * giststate->tupdesc->natts ); 
 	gistSplitByKey(r, page, itup, len, giststate, 
 		&v, entryvec, 0);
 
@@ -1073,31 +931,31 @@ gistSplit(Relation r,
 	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.splitVector.spl_nleft; i++)
+		lvectup[i] = itup[v.splitVector.spl_left[i] - 1];
 
-	for (i = 0; i < v.spl_nright; i++)
-		rvectup[i] = itup[v.spl_right[i] - 1];
+	for (i = 0; i < v.splitVector.spl_nright; i++)
+		rvectup[i] = itup[v.splitVector.spl_right[i] - 1];
 
 	/* finalyze splitting (may need another split) */
-	if (!gistfitpage(rvectup, v.spl_nright))
+	if (!gistfitpage(rvectup, v.splitVector.spl_nright))
 	{
-		res = gistSplit(r, page, rvectup, v.spl_nright, giststate);
+		res = gistSplit(r, page, rvectup, v.splitVector.spl_nright, giststate);
 	}
 	else
 	{
 		ROTATEDIST(res);
-		res->block.num = v.spl_nright;
-		res->list = gistfillitupvec(rvectup, v.spl_nright, &( res->lenlist ) );
-		res->itup = (v.spl_rightvalid) ? gistFormTuple(giststate, r, v.spl_rattr, v.spl_rattrsize, v.spl_risnull)
+		res->block.num = v.splitVector.spl_nright;
+		res->list = gistfillitupvec(rvectup, v.splitVector.spl_nright, &( res->lenlist ) );
+		res->itup = (v.spl_rightvalid) ? gistFormTuple(giststate, r, v.spl_rattr, v.spl_risnull, false)
 			: gist_form_invalid_tuple(GIST_ROOT_BLKNO);
 	}
 
-	if (!gistfitpage(lvectup, v.spl_nleft))
+	if (!gistfitpage(lvectup, v.splitVector.spl_nleft))
 	{
 		SplitedPageLayout *resptr, *subres;
 
-		resptr = subres = gistSplit(r, page, lvectup, v.spl_nleft, giststate);
+		resptr = subres = gistSplit(r, page, lvectup, v.splitVector.spl_nleft, giststate);
 
 		/* install on list's tail */ 
 		while( resptr->next )
@@ -1109,9 +967,9 @@ gistSplit(Relation r,
 	else
 	{
 		ROTATEDIST(res);
-		res->block.num = v.spl_nleft;
-		res->list = gistfillitupvec(lvectup, v.spl_nleft, &( res->lenlist ) );
-		res->itup = (v.spl_leftvalid) ? gistFormTuple(giststate, r, v.spl_lattr, v.spl_lattrsize, v.spl_lisnull)
+		res->block.num = v.splitVector.spl_nleft;
+		res->list = gistfillitupvec(lvectup, v.splitVector.spl_nleft, &( res->lenlist ) );
+		res->itup = (v.spl_leftvalid) ? gistFormTuple(giststate, r, v.spl_lattr, v.spl_lisnull, false)
 			: gist_form_invalid_tuple(GIST_ROOT_BLKNO);
 	}
 

src/backend/access/gist/gistget.c

@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/gist/gistget.c,v 1.57 2006/05/24 11:01:39 teodor Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/gist/gistget.c,v 1.58 2006/06/28 12:00:14 teodor Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -391,7 +391,6 @@ gistindex_keytest(IndexTuple tuple,
 
 		gistdentryinit(giststate, key->sk_attno - 1, &de,
 					   datum, r, p, offset,
-					   IndexTupleSize(tuple) - sizeof(IndexTupleData),
 					   FALSE, isNull);
 
 		/*

src/backend/access/gist/gistproc.c

@@ -10,7 +10,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	$PostgreSQL: pgsql/src/backend/access/gist/gistproc.c,v 1.5 2006/03/05 15:58:20 momjian Exp $
+ *	$PostgreSQL: pgsql/src/backend/access/gist/gistproc.c,v 1.6 2006/06/28 12:00:14 teodor Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -112,6 +112,17 @@ gist_box_consistent(PG_FUNCTION_ARGS)
 												 strategy));
 }
 
+static void
+adjustBox( BOX *b, BOX *addon ) {
+	if (b->high.x < addon->high.x)
+		b->high.x = addon->high.x;
+	if (b->low.x > addon->low.x)
+		b->low.x = addon->low.x;
+	if (b->high.y < addon->high.y)
+		b->high.y = addon->high.y;
+	if (b->low.y > addon->low.y)
+		b->low.y = addon->low.y;
+}
 
 /*
  * The GiST Union method for boxes
@@ -136,14 +147,7 @@ gist_box_union(PG_FUNCTION_ARGS)
 	for (i = 1; i < numranges; i++)
 	{
 		cur = DatumGetBoxP(entryvec->vector[i].key);
-		if (pageunion->high.x < cur->high.x)
-			pageunion->high.x = cur->high.x;
-		if (pageunion->low.x > cur->low.x)
-			pageunion->low.x = cur->low.x;
-		if (pageunion->high.y < cur->high.y)
-			pageunion->high.y = cur->high.y;
-		if (pageunion->low.y > cur->low.y)
-			pageunion->low.y = cur->low.y;
+		adjustBox( pageunion, cur );
 	}
 	*sizep = sizeof(BOX);
 
@@ -206,6 +210,74 @@ compare_KB(const void *a, const void *b)
 	return (sa > sb) ? 1 : -1;
 }
 
+static void
+chooseLR( GIST_SPLITVEC *v, 
+	OffsetNumber *list1, int nlist1, BOX *union1, 
+	OffsetNumber *list2, int nlist2, BOX *union2 )
+{
+	bool	firstToLeft = true;
+
+	if ( v->spl_ldatum_exists || v->spl_rdatum_exists ) {
+		if (  v->spl_ldatum_exists && v->spl_rdatum_exists ) {
+			BOX	LRl = *union1, LRr = *union2; 
+			BOX	RLl = *union2, RLr = *union1;
+			double sizeLR, sizeRL;
+
+			adjustBox( &LRl, DatumGetBoxP( v->spl_ldatum ) );
+			adjustBox( &LRr, DatumGetBoxP( v->spl_rdatum ) );
+			adjustBox( &RLl, DatumGetBoxP( v->spl_ldatum ) );
+			adjustBox( &RLr, DatumGetBoxP( v->spl_rdatum ) );
+
+			sizeLR = size_box( DirectFunctionCall2(rt_box_inter, BoxPGetDatum(&LRl), BoxPGetDatum(&LRr)) );
+			sizeRL = size_box( DirectFunctionCall2(rt_box_inter, BoxPGetDatum(&RLl), BoxPGetDatum(&RLr)) );
+
+			if ( sizeLR > sizeRL ) 
+				firstToLeft = false;
+
+		} else {
+			float p1, p2;
+			GISTENTRY	oldUnion, addon;
+
+			gistentryinit(oldUnion, ( v->spl_ldatum_exists ) ? v->spl_ldatum : v->spl_rdatum,
+						  NULL, NULL, InvalidOffsetNumber, FALSE);
+		
+			gistentryinit(addon, BoxPGetDatum(union1), NULL, NULL, InvalidOffsetNumber, FALSE);
+			DirectFunctionCall3(gist_box_penalty, PointerGetDatum(&oldUnion), PointerGetDatum(&union1), PointerGetDatum(&p1)); 
+			gistentryinit(addon, BoxPGetDatum(union2), NULL, NULL, InvalidOffsetNumber, FALSE);
+			DirectFunctionCall3(gist_box_penalty, PointerGetDatum(&oldUnion), PointerGetDatum(&union2), PointerGetDatum(&p2));
+
+			if (  (v->spl_ldatum_exists && p1 > p2) || (v->spl_rdatum_exists && p1 < p2)  ) 
+				firstToLeft = false;	
+		}
+	}
+
+	if ( firstToLeft ) {	
+		v->spl_left = list1;
+		v->spl_right = list2;
+		v->spl_nleft = nlist1;
+		v->spl_nright = nlist2;
+		if ( v->spl_ldatum_exists )
+			adjustBox(union1, DatumGetBoxP( v->spl_ldatum ) );
+		v->spl_ldatum = BoxPGetDatum(union1);
+		if ( v->spl_rdatum_exists )
+			adjustBox(union2, DatumGetBoxP( v->spl_rdatum ) );
+		v->spl_rdatum = BoxPGetDatum(union2);
+	} else {
+		v->spl_left = list2;
+		v->spl_right = list1;
+		v->spl_nleft = nlist2;
+		v->spl_nright = nlist1;
+		if ( v->spl_ldatum_exists )
+			adjustBox(union2, DatumGetBoxP( v->spl_ldatum ) );
+		v->spl_ldatum = BoxPGetDatum(union2);
+		if ( v->spl_rdatum_exists )
+			adjustBox(union1, DatumGetBoxP( v->spl_rdatum ) );
+		v->spl_rdatum = BoxPGetDatum(union1);
+	}
+
+	v->spl_ldatum_exists = v->spl_rdatum_exists = false;
+}
+
 /*
  * The GiST PickSplit method
  *
@@ -255,14 +327,7 @@ gist_box_picksplit(PG_FUNCTION_ARGS)
 								   ))
 			allisequal = false;
 
-		if (pageunion.high.x < cur->high.x)
-			pageunion.high.x = cur->high.x;
-		if (pageunion.low.x > cur->low.x)
-			pageunion.low.x = cur->low.x;
-		if (pageunion.high.y < cur->high.y)
-			pageunion.high.y = cur->high.y;
-		if (pageunion.low.y > cur->low.y)
-			pageunion.low.y = cur->low.y;
+		adjustBox( &pageunion, cur );
 	}
 
 	nbytes = (maxoff + 2) * sizeof(OffsetNumber);
@@ -294,9 +359,17 @@ gist_box_picksplit(PG_FUNCTION_ARGS)
 					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);
 		}
 	}
@@ -399,23 +472,13 @@ gist_box_picksplit(PG_FUNCTION_ARGS)
 	}
 
 	if (direction == 'x')
-	{
-		v->spl_left = listL;
-		v->spl_right = listR;
-		v->spl_nleft = posL;
-		v->spl_nright = posR;
-		v->spl_ldatum = BoxPGetDatum(unionL);
-		v->spl_rdatum = BoxPGetDatum(unionR);
-	}
-	else
-	{
-		v->spl_left = listB;
-		v->spl_right = listT;
-		v->spl_nleft = posB;
-		v->spl_nright = posT;
-		v->spl_ldatum = BoxPGetDatum(unionB);
-		v->spl_rdatum = BoxPGetDatum(unionT);
-	}
+		chooseLR( v,
+					listL, posL, unionL,
+					listR, posR, unionR );
+	else 
+		chooseLR( v,
+					listB, posB, unionB,
+					listT, posT, unionT );
 
 	PG_RETURN_POINTER(v);
 }
@@ -629,14 +692,14 @@ gist_poly_compress(PG_FUNCTION_ARGS)
 			memcpy((void *) r, (void *) &(in->boundbox), sizeof(BOX));
 			gistentryinit(*retval, PointerGetDatum(r),
 						  entry->rel, entry->page,
-						  entry->offset, sizeof(BOX), FALSE);
+						  entry->offset, FALSE);
 
 		}
 		else
 		{
 			gistentryinit(*retval, (Datum) 0,
 						  entry->rel, entry->page,
-						  entry->offset, 0, FALSE);
+						  entry->offset, FALSE);
 		}
 	}
 	else
@@ -700,14 +763,14 @@ gist_circle_compress(PG_FUNCTION_ARGS)
 			r->low.y = in->center.y - in->radius;
 			gistentryinit(*retval, PointerGetDatum(r),
 						  entry->rel, entry->page,
-						  entry->offset, sizeof(BOX), FALSE);
+						  entry->offset, FALSE);
 
 		}
 		else
 		{
 			gistentryinit(*retval, (Datum) 0,
 						  entry->rel, entry->page,
-						  entry->offset, 0, FALSE);
+						  entry->offset, FALSE);
 		}
 	}
 	else

src/include/access/gist.h

@@ -9,7 +9,7 @@
  * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/access/gist.h,v 1.53 2006/06/25 01:02:12 momjian Exp $
+ * $PostgreSQL: pgsql/src/include/access/gist.h,v 1.54 2006/06/28 12:00:14 teodor Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -75,37 +75,29 @@ typedef GISTPageOpaqueData *GISTPageOpaque;
 
 /*
  * This is the Split Vector to be returned by the PickSplit method.
+ * PickSplit should check spl_(r|l)datum_exists. If it is 'true',
+ * that corresponding spl_(r|l)datum already defined and
+ * PickSplit should use that value. PickSplit should always set
+ * spl_(r|l)datum_exists to false: GiST will check value to 
+ * control supportng this feature by PickSplit...
  */
 typedef struct GIST_SPLITVEC
 {
 	OffsetNumber *spl_left;		/* array of entries that go left */
 	int			spl_nleft;		/* size of this array */
 	Datum		spl_ldatum;		/* Union of keys in spl_left */
-	Datum		spl_lattr[INDEX_MAX_KEYS];		/* Union of subkeys in
-												 * spl_left */
-	int			spl_lattrsize[INDEX_MAX_KEYS];
-	bool		spl_lisnull[INDEX_MAX_KEYS];
-	bool		spl_leftvalid;
+	bool		spl_ldatum_exists;  /* true, if spl_ldatum already exists. */
 
 	OffsetNumber *spl_right;	/* array of entries that go right */
 	int			spl_nright;		/* size of the array */
 	Datum		spl_rdatum;		/* Union of keys in spl_right */
-	Datum		spl_rattr[INDEX_MAX_KEYS];		/* Union of subkeys in
-												 * spl_right */
-	int			spl_rattrsize[INDEX_MAX_KEYS];
-	bool		spl_risnull[INDEX_MAX_KEYS];
-	bool		spl_rightvalid;
-
-	int		   *spl_idgrp;
-	int		   *spl_ngrp;		/* number in each group */
-	char	   *spl_grpflag;	/* flags of each group */
+	bool		spl_rdatum_exists;  /* true, if spl_rdatum already exists. */
 } GIST_SPLITVEC;
 
 /*
  * An entry on a GiST node.  Contains the key, as well as its own
  * location (rel,page,offset) which can supply the matching pointer.
- * The size of the key is in bytes, and leafkey is a flag to tell us
- * if the entry is in a leaf node.
+ * leafkey is a flag to tell us if the entry is in a leaf node.
  */
 typedef struct GISTENTRY
 {
@@ -113,7 +105,6 @@ typedef struct GISTENTRY
 	Relation	rel;
 	Page		page;
 	OffsetNumber offset;
-	int			bytes;
 	bool		leafkey;
 } GISTENTRY;
 
@@ -147,8 +138,8 @@ typedef struct
 /*
  * macro to initialize a GISTENTRY
  */
-#define gistentryinit(e, k, r, pg, o, b, l) \
+#define gistentryinit(e, k, r, pg, o, l) \
 	do { (e).key = (k); (e).rel = (r); (e).page = (pg); \
-		 (e).offset = (o); (e).bytes = (b); (e).leafkey = (l); } while (0)
+		 (e).offset = (o); (e).leafkey = (l); } while (0)
 
 #endif   /* GIST_H */

src/include/access/gist_private.h

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/access/gist_private.h,v 1.17 2006/05/29 12:50:06 teodor Exp $
+ * $PostgreSQL: pgsql/src/include/access/gist_private.h,v 1.18 2006/06/28 12:00:14 teodor Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -180,6 +180,21 @@ typedef struct GISTInsertStack
 	struct GISTInsertStack *next;
 } GISTInsertStack;
 
+typedef struct GistSplitVector {
+	GIST_SPLITVEC	splitVector;  /* to/from PickSplit method */
+
+	Datum       spl_lattr[INDEX_MAX_KEYS];      /* Union of subkeys in spl_left */
+	bool        spl_lisnull[INDEX_MAX_KEYS];
+	bool        spl_leftvalid;
+
+	Datum       spl_rattr[INDEX_MAX_KEYS];      /* Union of subkeys in spl_right */
+	bool        spl_risnull[INDEX_MAX_KEYS];
+	bool        spl_rightvalid;
+
+	bool        *spl_equiv; /* equivalent tuples which can be freely 
+							 * distributed between left and right pages */ 
+} GistSplitVector;
+
 #define XLogRecPtrIsInvalid( r )	( (r).xlogid == 0 && (r).xrecoff == 0 )
 
 typedef struct
@@ -206,12 +221,6 @@ typedef struct
 /* root page of a gist index */
 #define GIST_ROOT_BLKNO				0
 
-#define ATTSIZE(datum, tupdesc, i, isnull) \
-		( \
-				(isnull) ? 0 : \
-				   att_addlength(0, (tupdesc)->attrs[(i)-1]->attlen, (datum)) \
-		)
-
 /*
  * mark tuples on inner pages during recovery
  */
@@ -281,7 +290,7 @@ extern IndexTuple gistgetadjusted(Relation r,
 				IndexTuple addtup,
 				GISTSTATE *giststate);
 extern IndexTuple gistFormTuple(GISTSTATE *giststate,
-			  Relation r, Datum *attdata, int *datumsize, bool *isnull);
+			  Relation r, Datum *attdata, bool *isnull, bool newValues);
 
 extern OffsetNumber gistchoose(Relation r, Page p,
 		   IndexTuple it,
@@ -289,34 +298,34 @@ extern OffsetNumber gistchoose(Relation r, Page p,
 extern void gistcentryinit(GISTSTATE *giststate, int nkey,
 			   GISTENTRY *e, Datum k,
 			   Relation r, Page pg,
-			   OffsetNumber o, int b, bool l, bool isNull);
-extern void gistDeCompressAtt(GISTSTATE *giststate, Relation r,
-				  IndexTuple tuple, Page p, OffsetNumber o,
-				  GISTENTRY *attdata, bool *isnull);
-
-typedef struct {
-	int		*attrsize;
-	Datum	*attr;
-	int		len;
-	OffsetNumber *entries;
-	bool	*isnull;
-	int		*idgrp;
-} GistSplitVec;
-
-extern void gistunionsubkeyvec(GISTSTATE *giststate, 
-	IndexTuple *itvec, GistSplitVec *gsvp,  int startkey);
-extern void gistunionsubkey(GISTSTATE *giststate, IndexTuple *itvec, 
-		GIST_SPLITVEC *spl, int attno);
+			   OffsetNumber o, bool l, bool isNull);
 
 extern void GISTInitBuffer(Buffer b, uint32 f);
 extern void gistdentryinit(GISTSTATE *giststate, int nkey, GISTENTRY *e,
 			   Datum k, Relation r, Page pg, OffsetNumber o,
-			   int b, bool l, bool isNull);
-bool gistUserPicksplit(Relation r, GistEntryVector *entryvec, int attno, GIST_SPLITVEC *v,
-				  IndexTuple *itup, int len, GISTSTATE *giststate);
+			   bool l, bool isNull);
+
+extern float gistpenalty(GISTSTATE *giststate, int attno,
+            			GISTENTRY *key1, bool isNull1,
+						GISTENTRY *key2, bool isNull2);
+extern bool gistMakeUnionItVec(GISTSTATE *giststate, IndexTuple *itvec, int len, int startkey,
+                    Datum *attr, bool *isnull );
+extern bool gistKeyIsEQ(GISTSTATE *giststate, int attno, Datum a, Datum b);
+extern void gistDeCompressAtt(GISTSTATE *giststate, Relation r, IndexTuple tuple, Page p,
+				OffsetNumber o, GISTENTRY *attdata, bool *isnull);
+
+extern void gistMakeUnionKey( GISTSTATE *giststate, int attno,
+                    GISTENTRY   *entry1, bool isnull1,
+					GISTENTRY   *entry2, bool isnull2,
+					Datum   *dst, bool *dstisnull );
 
 /* gistvacuum.c */
 extern Datum gistbulkdelete(PG_FUNCTION_ARGS);
 extern Datum gistvacuumcleanup(PG_FUNCTION_ARGS);
 
+/* gistsplit.c */
+extern void gistSplitByKey(Relation r, Page page, IndexTuple *itup, 
+							int len, GISTSTATE *giststate,
+							GistSplitVector *v, GistEntryVector *entryvec, 
+							int attno);
 #endif   /* GIST_PRIVATE_H */