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

from buggy user-defined picksplit to GiST.

diff --git a/src/backend/access/gist/gistproc.c b/src/backend/access/gist/gistproc.c
index df9ecbbb2f386e621985a9463692c6813381d0fd..c899da2128c291f82c370259a7f8ddb3d801739b 100644 (file)
--- a/src/backend/access/gist/gistproc.c
+++ b/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.3.2.1 2007/09/07 17:19:52 teodor Exp $
+ *     $PostgreSQL: pgsql/src/backend/access/gist/gistproc.c,v 1.3.2.2 2009/04/07 17:46:37 teodor Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -142,6 +142,71 @@ gist_box_penalty(PG_FUNCTION_ARGS)
        PG_RETURN_POINTER(result);
 }
 
+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;
+}
+
+/*
+ * 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++;
+               }
+       }
+}
+
 /*
  * The GiST PickSplit method
  *
@@ -191,54 +256,25 @@ 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);
-       listL = (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++;
-                               }
-                       }
-                       v->spl_ldatum = BoxPGetDatum(unionL);
-                       v->spl_rdatum = BoxPGetDatum(unionR);
-
-                       PG_RETURN_POINTER(v);
-               }
+               /*
+                * All entries are the same
+                */
+               fallbackSplit(entryvec, v);
+               PG_RETURN_POINTER(v);
        }
 
+       nbytes = (maxoff + 2) * sizeof(OffsetNumber);
+       listL = (OffsetNumber *) palloc(nbytes);
+       listR = (OffsetNumber *) palloc(nbytes);
        listB = (OffsetNumber *) palloc(nbytes);
        listT = (OffsetNumber *) palloc(nbytes);
+       unionL = (BOX *) palloc(sizeof(BOX));
+       unionR = (BOX *) palloc(sizeof(BOX));
        unionB = (BOX *) palloc(sizeof(BOX));
        unionT = (BOX *) palloc(sizeof(BOX));
 
@@ -319,6 +355,12 @@ gist_box_picksplit(PG_FUNCTION_ARGS)
                        else 
                                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? */

diff --git a/src/backend/access/gist/gistutil.c b/src/backend/access/gist/gistutil.c
index 16b1ceecf28a6d9f1de29828f7a9ceba07c49758..db12cf96fa8f4bf5e1ef0b892146cb71eb2f1d40 100644 (file)
--- a/src/backend/access/gist/gistutil.c
+++ b/src/backend/access/gist/gistutil.c
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *                     $PostgreSQL: pgsql/src/backend/access/gist/gistutil.c,v 1.7 2005/09/22 20:44:36 momjian Exp $
+ *                     $PostgreSQL: pgsql/src/backend/access/gist/gistutil.c,v 1.7.2.1 2009/04/07 17:46:37 teodor Exp $
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
@@ -815,6 +815,63 @@ GISTInitBuffer(Buffer b, uint32 f)
        memset(&(opaque->nsn), 0, sizeof(GistNSN));
 }
 
+/*
+ * 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)
+{
+       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[0],
+                                                                       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[0],
+                                                                       PointerGetDatum(evec),
+                                                                       PointerGetDatum(&nbytes));
+}
+
 void
 gistUserPicksplit(Relation r, GistEntryVector *entryvec, GIST_SPLITVEC *v,
                                  IndexTuple *itup, int len, GISTSTATE *giststate)
@@ -827,11 +884,24 @@ gistUserPicksplit(Relation r, GistEntryVector *entryvec, GIST_SPLITVEC *v,
                                  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) (entryvec->n - 1);
-       if (v->spl_right[v->spl_nright - 1] == InvalidOffsetNumber)
-               v->spl_right[v->spl_nright - 1] = (OffsetNumber) (entryvec->n - 1);
+       if ( v->spl_nleft == 0 || v->spl_nright == 0 )
+       {
+               ereport(DEBUG1,
+                               (errcode(ERRCODE_INTERNAL_ERROR),
+                                errmsg("Picksplit method for first column of index \"%s\" failed",
+                                                                                               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")));
+
+               genericPickSplit(giststate, entryvec, v);
+       }
+       else
+       {
+               /* compatibility with old code */
+               if (v->spl_left[v->spl_nleft - 1] == InvalidOffsetNumber)
+                       v->spl_left[v->spl_nleft - 1] = (OffsetNumber) (entryvec->n - 1);
+               if (v->spl_right[v->spl_nright - 1] == InvalidOffsetNumber)
+                       v->spl_right[v->spl_nright - 1] = (OffsetNumber) (entryvec->n - 1);
+       }
 
        v->spl_lattr[0] = v->spl_ldatum;
        v->spl_rattr[0] = v->spl_rdatum;