GIN index build speed

While playing around with the "GIN fast insert" patch, I was puzzled why
building a GIN index seemed to take so long. The test case I use was simply:

CREATE TABLE foo (bar tsvector);
INSERT INTO foo SELECT to_tsvector('foo' || a) FROM generate_series(1,
200000) a;
CREATE INDEX foogin ON foo USING gin (bar);

The CREATE INDEX step takes about 40 seconds on my laptop, which seems
excessive.

The issue is that the GIN index build code accumulates the lexemes into
a binary tree, but there's nothing to keep the tree balanced. My test
case with almost monotonically increasing keys, happens to be a
worst-case scenario, and the tree degenerates into almost linked list
that every insertion has to grovel through.

The obvious fix is to use a balanced tree algorithm. I wrote a quick
patch to turn the tree into a splay tree. That fixed the degenerative
behavior, and the runtime of CREATE INDEX for the above test case fell
from 40s to 1.5s.

Magnus kindly gave me a dump of the full-text-search tables from
search.postgresql.org, for some real-world testing. Quick testing with
that suggests that the patch unfortunately makes the index build 5-10%
slower with that data set.

We're in commitfest, not supposed to be submitting new features, so I'm
not going to pursue this further right now. Patch attached, however,
which seems to work fine.

--
   Heikki Linnakangas
   EnterpriseDB   http://www.enterprisedb.com
*** src/backend/access/gin/ginbulk.c
--- src/backend/access/gin/ginbulk.c
***************
*** 25,31 ****
  void
  ginInitBA(BuildAccumulator *accum)
  {
!     accum->maxdepth = 1;
      accum->stackpos = 0;
      accum->entries = NULL;
      accum->stack = NULL;
--- 25,31 ----
  void
  ginInitBA(BuildAccumulator *accum)
  {
!     accum->stacksize = 0;
      accum->stackpos = 0;
      accum->entries = NULL;
      accum->stack = NULL;
***************
*** 98,159 **** getDatumCopy(BuildAccumulator *accum, OffsetNumber attnum, Datum value)
  }

  /*
   * Find/store one entry from indexed value.
   */
  static void
  ginInsertEntry(BuildAccumulator *accum, ItemPointer heapptr, OffsetNumber attnum, Datum entry)
  {
!     EntryAccumulator *ea = accum->entries,
!                *pea = NULL;
      int            res = 0;
-     uint32        depth = 1;

!     while (ea)
      {
!         res = compareAttEntries(accum->ginstate, attnum, entry, ea->attnum, ea->value);
          if (res == 0)
-             break;                /* found */
-         else
          {
!             pea = ea;
!             if (res < 0)
!                 ea = ea->left;
!             else
!                 ea = ea->right;
          }
-         depth++;
      }

!     if (depth > accum->maxdepth)
!         accum->maxdepth = depth;

!     if (ea == NULL)
!     {
!         ea = EAAllocate(accum);

          ea->left = ea->right = NULL;
!         ea->attnum = attnum;
!         ea->value = getDatumCopy(accum, attnum, entry);
!         ea->length = DEF_NPTR;
!         ea->number = 1;
!         ea->shouldSort = FALSE;
!         ea->list = (ItemPointerData *) palloc(sizeof(ItemPointerData) * DEF_NPTR);
!         accum->allocatedMemory += GetMemoryChunkSpace(ea->list);
!         ea->list[0] = *heapptr;
!
!         if (pea == NULL)
!             accum->entries = ea;
          else
          {
!             Assert(res != 0);
!             if (res < 0)
!                 pea->left = ea;
!             else
!                 pea->right = ea;
          }
      }
!     else
!         ginInsertData(accum, ea, heapptr);
  }

  /*
--- 98,234 ----
  }

  /*
+  * Adapted to C from the Public Domain splay tree implementation at:
+  * http://www.link.cs.cmu.edu/link/ftp-site/splaying/SplayTree.java
+  * Original implementation by Danny Sleator <sleator@cs.cmu.edu>
+  *
+  * Internal method to perform a top-down splay.
+  *
+  *   splay(key) does the splay operation on the given key.
+  *   If key is in the tree, then the BinaryNode containing
+  *   that key becomes the root.  If key is not in the tree,
+  *   then after the splay, key.root is either the greatest key
+  *   < key in the tree, or the lest key > key in the tree.
+  *
+  *   This means, among other things, that if you splay with
+  *   a key that's larger than any in the tree, the rightmost
+  *   node of the tree becomes the root.  This property is used
+  *   in the delete() method.
+  */
+ static void
+ splay(BuildAccumulator *accum, OffsetNumber key_attnum, Datum key_value)
+ {
+     EntryAccumulator *l, *r, *t, *y;
+     EntryAccumulator header;
+     int res;
+
+     l = r = &header;
+     t = accum->entries;
+     header.left = header.right = NULL;
+     for (;;)
+     {
+         res = compareAttEntries(accum->ginstate, key_attnum, key_value,
+                                 t->attnum, t->value);
+         if (res < 0)
+         {
+             if (t->left == NULL) break;
+             if (compareAttEntries(accum->ginstate, key_attnum, key_value,
+                                   t->left->attnum, t->left->value) < 0)
+             {
+                 y = t->left;                            /* rotate right */
+                 t->left = y->right;
+                 y->right = t;
+                 t = y;
+                 if (t->left == NULL) break;
+             }
+             r->left = t;                                 /* link right */
+             r = t;
+             t = t->left;
+         }
+         else if (res > 0)
+         {
+             if (t->right == NULL) break;
+             if (compareAttEntries(accum->ginstate, key_attnum, key_value,
+                                   t->right->attnum, t->right->value) > 0)
+             {
+                 y = t->right;                            /* rotate left */
+                 t->right = y->left;
+                 y->left = t;
+                 t = y;
+                 if (t->right == NULL) break;
+             }
+             l->right = t;                                /* link left */
+             l = t;
+             t = t->right;
+         }
+         else
+         {
+             break;
+         }
+     }
+     l->right = t->left;                                   /* assemble */
+     r->left = t->right;
+     t->left = header.right;
+     t->right = header.left;
+     accum->entries = t;
+ }
+
+ /*
   * Find/store one entry from indexed value.
   */
  static void
  ginInsertEntry(BuildAccumulator *accum, ItemPointer heapptr, OffsetNumber attnum, Datum entry)
  {
!     EntryAccumulator *root = accum->entries;
!     EntryAccumulator *ea;
      int            res = 0;

!     if (root != NULL)
      {
!         splay(accum, attnum, entry);
!         root = accum->entries;
!
!         res = compareAttEntries(accum->ginstate, attnum, entry,
!                                 root->attnum, root->value);
          if (res == 0)
          {
!             /* found */
!             ginInsertData(accum, root, heapptr);
!             return;
          }
      }

!     /* Not found. Insert */

!     ea = EAAllocate(accum);
!
!     ea->attnum = attnum;
!     ea->value = getDatumCopy(accum, attnum, entry);
!     ea->length = DEF_NPTR;
!     ea->number = 1;
!     ea->shouldSort = FALSE;
!     ea->list = (ItemPointerData *) palloc(sizeof(ItemPointerData) * DEF_NPTR);
!     accum->allocatedMemory += GetMemoryChunkSpace(ea->list);
!     ea->list[0] = *heapptr;

+     if (root == NULL)
          ea->left = ea->right = NULL;
!     else
!     {
!         if (res < 0)
!         {
!             ea->left = root->left;
!             ea->right = root;
!             root->left = NULL;
!         }
          else
          {
!             ea->right = root->right;
!             ea->left = root;
!             root->right = NULL;
          }
      }
!     accum->entries = ea;
  }

  /*
***************
*** 217,222 **** qsortCompareItemPointers(const void *a, const void *b)
--- 292,314 ----
      return res;
  }

+ /* Push an entry to accum->stack */
+ static void
+ pushEntry(BuildAccumulator *accum, EntryAccumulator *e)
+ {
+     accum->stack[++accum->stackpos] = e;
+
+     if (accum->stackpos + 1 == accum->stacksize)
+     {
+         accum->stacksize++;
+         accum->allocatedMemory -= GetMemoryChunkSpace(accum->stack);
+         accum->stack = repalloc(accum->stack,
+                                 sizeof(EntryAccumulator *) * accum->stacksize);
+         accum->allocatedMemory += GetMemoryChunkSpace(accum->stack);
+         accum->stack[accum->stacksize - 1] = NULL;
+     }
+ }
+
  /*
   * walk on binary tree and returns ordered nodes
   */
***************
*** 230,252 **** walkTree(BuildAccumulator *accum)
          /* return entry itself: we already was at left sublink */
          return entry;
      }
!     else if (entry->right && entry->right != accum->stack[accum->stackpos + 1])
      {
          /* go on right sublink */
-         accum->stackpos++;
          entry = entry->right;

          /* find most-left value */
!         for (;;)
          {
!             accum->stack[accum->stackpos] = entry;
!             if (entry->left)
!             {
!                 accum->stackpos++;
!                 entry = entry->left;
!             }
!             else
!                 break;
          }
      }
      else
--- 322,339 ----
          /* return entry itself: we already was at left sublink */
          return entry;
      }
!     else if (entry->right &&
!              entry->right != accum->stack[accum->stackpos + 1])
      {
          /* go on right sublink */
          entry = entry->right;
+         pushEntry(accum, entry);

          /* find most-left value */
!         while(entry->left)
          {
!             entry = entry->left;
!             pushEntry(accum, entry);
          }
      }
      else
***************
*** 270,293 **** ginGetEntry(BuildAccumulator *accum, OffsetNumber *attnum, Datum *value, uint32
      if (accum->stack == NULL)
      {
          /* first call */
!         accum->stack = palloc0(sizeof(EntryAccumulator *) * (accum->maxdepth + 1));
          accum->allocatedMemory += GetMemoryChunkSpace(accum->stack);
          entry = accum->entries;

          if (entry == NULL)
              return NULL;

          /* find most-left value */
!         for (;;)
          {
!             accum->stack[accum->stackpos] = entry;
!             if (entry->left)
!             {
!                 accum->stackpos++;
!                 entry = entry->left;
!             }
!             else
!                 break;
          }
      }
      else
--- 357,376 ----
      if (accum->stack == NULL)
      {
          /* first call */
!         accum->stacksize = 10;
!         accum->stack = palloc0(sizeof(EntryAccumulator *) * accum->stacksize);
          accum->allocatedMemory += GetMemoryChunkSpace(accum->stack);
          entry = accum->entries;

          if (entry == NULL)
              return NULL;

+         accum->stack[0] = entry;
          /* find most-left value */
!         while(entry->left)
          {
!             entry = entry->left;
!             pushEntry(accum, entry);
          }
      }
      else
*** src/include/access/gin.h
--- src/include/access/gin.h
***************
*** 473,479 **** typedef struct
  {
      GinState   *ginstate;
      EntryAccumulator *entries;
!     uint32        maxdepth;
      EntryAccumulator **stack;
      uint32        stackpos;
      long        allocatedMemory;
--- 473,479 ----
  {
      GinState   *ginstate;
      EntryAccumulator *entries;
!     int            stacksize;
      EntryAccumulator **stack;
      uint32        stackpos;
      long        allocatedMemory;