Fix duplicating ROOT page in concurrent updates.

diff --git a/src/backend/access/nbtree/nbtinsert.c b/src/backend/access/nbtree/nbtinsert.c
index a9d4508e58639f9c5d0476d30ae56ae39f8c4990..0daa91857dbe4c8e3be4cb442553492bb0836137 100644 (file)
--- a/src/backend/access/nbtree/nbtinsert.c
+++ b/src/backend/access/nbtree/nbtinsert.c
@@ -398,6 +398,7 @@ _bt_insertonpg(Relation rel,
                OffsetNumber maxoff;
                bool            shifted = false;
                bool            left_chained = (lpageop->btpo_flags & BTP_CHAIN) ? true : false;
+               bool            is_root = lpageop->btpo_flags & BTP_ROOT;
 
                /*
                 * If we have to split leaf page in the chain of duplicates by new
@@ -570,9 +571,20 @@ _bt_insertonpg(Relation rel,
                 * reasoning).
                 */
 
+l_spl:;
                if (stack == (BTStack) NULL)
                {
-
+                       if (!is_root)   /* if this page was not root page */
+                       {
+                               elog(DEBUG, "btree: concurrent ROOT page split");
+                               stack = (BTStack) palloc(sizeof(BTStackData));
+                               stack->bts_blkno = lpageop->btpo_parent;
+                               stack->bts_offset = InvalidOffsetNumber;
+                               stack->bts_btitem = (BTItem) palloc(sizeof(BTItemData));
+                               /* bts_btitem will be initialized below */
+                               stack->bts_parent = NULL;
+                               goto l_spl;
+                       }
                        /* create a new root node and release the split buffers */
                        _bt_newroot(rel, buf, rbuf);
                        _bt_relbuf(rel, buf, BT_WRITE);

diff --git a/src/backend/access/nbtree/nbtpage.c b/src/backend/access/nbtree/nbtpage.c
index 7c7ef33db00cf2ad3f052ca1dec8b260fd5ecff8..6ec8d1e654b56cda08910e8d62665f1e40947659 100644 (file)
--- a/src/backend/access/nbtree/nbtpage.c
+++ b/src/backend/access/nbtree/nbtpage.c
@@ -494,24 +494,37 @@ _bt_getstackbuf(Relation rel, BTStack stack, int access)
        opaque = (BTPageOpaque) PageGetSpecialPointer(page);
        maxoff = PageGetMaxOffsetNumber(page);
 
-       if (maxoff >= stack->bts_offset)
+       if (stack->bts_offset == InvalidOffsetNumber || 
+               maxoff >= stack->bts_offset)
        {
-               itemid = PageGetItemId(page, stack->bts_offset);
-               item = (BTItem) PageGetItem(page, itemid);
-
-               /* if the item is where we left it, we're done */
-               if (BTItemSame(item, stack->bts_btitem))
+               /*
+                * _bt_insertonpg set bts_offset to InvalidOffsetNumber
+                * in the case of concurrent ROOT page split
+                */
+               if (stack->bts_offset == InvalidOffsetNumber)
                {
-                       pfree(stack->bts_btitem);
-                       item_nbytes = ItemIdGetLength(itemid);
-                       item_save = (BTItem) palloc(item_nbytes);
-                       memmove((char *) item_save, (char *) item, item_nbytes);
-                       stack->bts_btitem = item_save;
-                       return buf;
+                       i = P_RIGHTMOST(opaque) ? P_HIKEY : P_FIRSTKEY;
+               }
+               else
+               {
+                       itemid = PageGetItemId(page, stack->bts_offset);
+                       item = (BTItem) PageGetItem(page, itemid);
+
+                       /* if the item is where we left it, we're done */
+                       if (BTItemSame(item, stack->bts_btitem))
+                       {
+                               pfree(stack->bts_btitem);
+                               item_nbytes = ItemIdGetLength(itemid);
+                               item_save = (BTItem) palloc(item_nbytes);
+                               memmove((char *) item_save, (char *) item, item_nbytes);
+                               stack->bts_btitem = item_save;
+                               return buf;
+                       }
+                       i = OffsetNumberNext(stack->bts_offset);
                }
 
                /* if the item has just moved right on this page, we're done */
-               for (i = OffsetNumberNext(stack->bts_offset);
+               for ( ;
                         i <= maxoff;
                         i = OffsetNumberNext(i))
                {