WAL

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 3e1de33bfe40e4649904386f2fa51a050c86f1a9..794ba977cc15793083111f2b06986921478464b8 100644 (file)
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $Header: /cvsroot/pgsql/src/backend/access/heap/heapam.c,v 1.87 2000/10/13 02:02:59 vadim Exp $
+ *       $Header: /cvsroot/pgsql/src/backend/access/heap/heapam.c,v 1.88 2000/10/13 12:05:20 vadim Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -2198,7 +2198,8 @@ void heap_xlog_insert(bool redo, XLogRecPtr lsn, XLogRecord *record)
        HeapTupleHeader htup = (HeapTupleHeader) PageGetItem(page, lp);
 
        /* is it our tuple ? */
-       if (htup->t_xmin != record->xl_xid || htup->t_cmin != xlrec->target.cid)
+       if (PageGetSUI(page) != ThisStartUpID || 
+               htup->t_xmin != record->xl_xid || htup->t_cmin != xlrec->target.cid)
        {
                if (!InRecovery)
                        elog(STOP, "heap_insert_undo: invalid target tuple in rollback");
@@ -2394,7 +2395,8 @@ newt:;
        htup = (HeapTupleHeader) PageGetItem(page, lp);
 
        /* is it our tuple ? */
-       if (htup->t_xmin != record->xl_xid || htup->t_cmin != xlrec->target.cid)
+       if (PageGetSUI(page) != ThisStartUpID || 
+               htup->t_xmin != record->xl_xid || htup->t_cmin != xlrec->target.cid)
        {
                if (!InRecovery)
                        elog(STOP, "heap_update_undo: invalid new tuple in rollback");

diff --git a/src/backend/access/nbtree/nbtinsert.c b/src/backend/access/nbtree/nbtinsert.c
index c72b8ca3df613289213f466596e65f81fe8773ee..0105739b5cd865459f0ebaca61115b38b2809ce7 100644 (file)
--- a/src/backend/access/nbtree/nbtinsert.c
+++ b/src/backend/access/nbtree/nbtinsert.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtinsert.c,v 1.65 2000/10/13 02:03:00 vadim Exp $
+ *       $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtinsert.c,v 1.66 2000/10/13 12:05:20 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -612,6 +612,10 @@ _bt_split(Relation rel, Buffer buf, OffsetNumber firstright,
        OffsetNumber maxoff;
        OffsetNumber i;
 
+#ifdef XLOG
+       BTItem          lhikey;
+#endif
+
        rbuf = _bt_getbuf(rel, P_NEW, BT_WRITE);
        origpage = BufferGetPage(buf);
        leftpage = PageGetTempPage(origpage, sizeof(BTPageOpaqueData));
@@ -680,6 +684,9 @@ _bt_split(Relation rel, Buffer buf, OffsetNumber firstright,
                itemsz = ItemIdGetLength(itemid);
                item = (BTItem) PageGetItem(origpage, itemid);
        }
+#ifdef XLOG
+       lhikey = item;
+#endif
        if (PageAddItem(leftpage, (Item) item, itemsz, leftoff,
                                        LP_USED) == InvalidOffsetNumber)
                elog(STOP, "btree: failed to add hikey to the left sibling");
@@ -793,12 +800,19 @@ _bt_split(Relation rel, Buffer buf, OffsetNumber firstright,
                        memcpy(xlbuf + hsize, &(_xlheapRel->rd_node), sizeof(RelFileNode));
                        hsize += sizeof(RelFileNode);
                }
+               else
+               {
+                       Size    itemsz = IndexTupleDSize(lhikey->bti_itup) + 
+                                               (sizeof(BTItemData) - sizeof(IndexTupleData));
+                       memcpy(xlbuf + hsize, (char*) lhikey, itemsz);
+                       hsize += itemsz;
+               }
                if (newitemonleft)
                {
                        /*
                         * Read comments in _bt_pgaddtup.
                         * Actually, seems that in non-leaf splits newitem shouldn't
-                        * go to first data key position.
+                        * go to first data key position on left page.
                         */
                        if (! P_ISLEAF(lopaque) && itup_off == P_FIRSTDATAKEY(lopaque))
                        {

diff --git a/src/backend/access/nbtree/nbtree.c b/src/backend/access/nbtree/nbtree.c
index 1064c2bb1075c7ff6c9f2b71227687569d9a6c17..a57bac1c810ee2d61dd6318f10fd9b690aabb127 100644 (file)
--- a/src/backend/access/nbtree/nbtree.c
+++ b/src/backend/access/nbtree/nbtree.c
@@ -12,7 +12,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *       $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtree.c,v 1.64 2000/10/13 02:03:00 vadim Exp $
+ *       $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtree.c,v 1.65 2000/10/13 12:05:20 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -837,7 +837,7 @@ static void btree_xlog_insert(bool redo, XLogRecPtr lsn, XLogRecord *record)
                                        ItemPointerGetOffsetNumber(&(xlrec->target.tid)),
                                        (char*)xlrec + hsize,
                                        record->xl_len - hsize,
-                                       &hnode))
+                                       hnode))
                                elog(STOP, "btree_insert_redo: failed to add item");
 
                        PageSetLSN(page, lsn);
@@ -908,7 +908,7 @@ btree_xlog_split(bool redo, bool onleft, XLogRecPtr lsn, XLogRecord *record)
                else
                {
                        /* Delete items related to new right sibling */
-                       _bt_thin_left_page(page, record);
+                       _bt_fix_left_page(page, record, onleft);
 
                        if (onleft)
                        {
@@ -924,6 +924,13 @@ btree_xlog_split(bool redo, bool onleft, XLogRecPtr lsn, XLogRecord *record)
                                        memcpy(&hnode, (char*)xlrec + SizeOfBtreeSplit + 
                                                                sizeof(CommandId), sizeof(RelFileNode));
                                }
+                               else
+                               {
+                                       memcpy(&btdata, (char*)xlrec + hsize, sizeof(BTItemData));
+                                       itemsz = IndexTupleDSize(btdata.bti_itup) +
+                                                               (sizeof(BTItemData) - sizeof(IndexTupleData));
+                                       hsize += itemsz;
+                               }
 
                                memcpy(&btdata, (char*)xlrec + hsize, sizeof(BTItemData));
                                itemsz = IndexTupleDSize(btdata.bti_itup) +
@@ -933,7 +940,7 @@ btree_xlog_split(bool redo, bool onleft, XLogRecPtr lsn, XLogRecord *record)
                                                ItemPointerGetOffsetNumber(&(xlrec->target.tid)),
                                                (char*)xlrec + hsize,
                                                itemsz,
-                                               &hnode))
+                                               hnode))
                                        elog(STOP, "btree_split_redo: failed to add item");
                        }
                        else
@@ -994,6 +1001,13 @@ btree_xlog_split(bool redo, bool onleft, XLogRecPtr lsn, XLogRecord *record)
                                pageop->btpo_flags |= BTP_LEAF;
                                hsize += (sizeof(CommandId) + sizeof(RelFileNode));
                        }
+                       else
+                       {
+                               memcpy(&btdata, (char*)xlrec + hsize, sizeof(BTItemData));
+                               itemsz = IndexTupleDSize(btdata.bti_itup) +
+                                                       (sizeof(BTItemData) - sizeof(IndexTupleData));
+                               hsize += itemsz;
+                       }
                        if (onleft)             /* skip target item */
                        {
                                memcpy(&btdata, (char*)xlrec + hsize, sizeof(BTItemData));
@@ -1198,17 +1212,28 @@ _bt_del_item(Relation reln, Buffer buffer, BTItem btitem, bool insert,
 {
        char               *xlrec = (char*) XLogRecGetData(record);
        Page                    page = (Page) BufferGetPage(buffer);
-       BTPageOpaque    pageop;
+       BTPageOpaque    pageop = (BTPageOpaque) PageGetSpecialPointer(page);
        BlockNumber             blkno;
        OffsetNumber    offno;
        ItemId                  lp;
+       BTItem                  item;
 
        for ( ; ; )
        {
-               offno = _bt_find_btitem(page, btitem);
-               if (offno != InvalidOffsetNumber)
+               OffsetNumber    maxoff = PageGetMaxOffsetNumber(page);
+
+               for (offno = P_FIRSTDATAKEY(pageop);
+                        offno <= maxoff;
+                        offno = OffsetNumberNext(offno))
+               {
+                       lp = PageGetItemId(page, offno);
+                       item = (BTItem) PageGetItem(page, lp);
+                       if (BTItemSame(item, btitem))
+                               break;
+               }
+               if (offno <= maxoff)
                        break;
-               pageop = (BTPageOpaque) PageGetSpecialPointer(page);
+               offno = InvalidOffsetNumber;
                if (P_RIGHTMOST(pageop))
                        break;
                blkno = pageop->btpo_next;
@@ -1221,6 +1246,7 @@ _bt_del_item(Relation reln, Buffer buffer, BTItem btitem, bool insert,
                if (PageIsNew((PageHeader) page))
                        elog(STOP, "btree_%s_undo: uninitialized right sibling",
                                (insert) ? "insert" : "split");
+               pageop = (BTPageOpaque) PageGetSpecialPointer(page);
                if (XLByteLT(PageGetLSN(page), lsn))
                        break;
        }
@@ -1250,9 +1276,9 @@ _bt_del_item(Relation reln, Buffer buffer, BTItem btitem, bool insert,
 
                memcpy(&cid, (char*)xlrec + hsize, sizeof(CommandId));
                memcpy(&hnode, (char*)xlrec + hsize + sizeof(CommandId), sizeof(RelFileNode));
-               result = XLogCheckHeapTuple(hnode, &(btitem->bti_itup.t_tid),
+               result = XLogIsOwnerOfTuple(hnode, &(btitem->bti_itup.t_tid),
                                        record->xl_xid, cid);
-               if (result <= 0)        /* no tuple or not owner */
+               if (result < 0) /* not owner */
                {
                        UnlockAndReleaseBuffer(buffer);
                        return;
@@ -1278,7 +1304,7 @@ _bt_del_item(Relation reln, Buffer buffer, BTItem btitem, bool insert,
 
 static bool
 _bt_add_item(Page page, OffsetNumber offno, 
-       char* item, Size size, RelFileNode* hnode)
+       char* item, Size size, RelFileNode hnode)
 {
        BTPageOpaque    pageop = (BTPageOpaque) PageGetSpecialPointer(page);
 
@@ -1309,4 +1335,125 @@ _bt_add_item(Page page, OffsetNumber offno,
        return(true);
 }
 
+static bool
+_bt_cleanup_page(Page page, RelFileNode hnode)
+{
+       OffsetNumber    maxoff = PageGetMaxOffsetNumber(page);
+       OffsetNumber    offno;
+       ItemId                  lp;
+       BTItem                  item;
+       bool                    result = false;
+
+       for (offno = P_FIRSTDATAKEY(pageop); offno <= maxoff; )
+       {
+               lp = PageGetItemId(page, offno);
+               item = (BTItem) PageGetItem(page, lp);
+               if (XLogIsValidTuple(hnode, &(item->bti_itup.t_tid))
+                       offno = OffsetNumberNext(offno);
+               else
+               {
+                       PageIndexTupleDelete(page, offno);
+                       maxoff = PageGetMaxOffsetNumber(page);
+                       result = true;
+               }
+       }
+
+       return(result);
+}
+
+/*
+ * Remove from left sibling items belonging to right sibling
+ * and change P_HIKEY
+ */
+static void
+_bt_fix_left_page(Page page, XLogRecord *record, bool onleft)
+{
+       char               *xlrec = (char*) XLogRecGetData(record);
+       BTPageOpaque    pageop = (BTPageOpaque) PageGetSpecialPointer(page);
+       Size                    hsize = SizeOfBtreeSplit;
+       RelFileNode             hnode;
+       BTItemData              btdata;
+       OffsetNumber    maxoff = PageGetMaxOffsetNumber(page);
+       OffsetNumber    offno;
+       char               *item;
+       Size                    itemsz;
+       char               *previtem = NULL;
+       char               *lhikey = NULL;
+       Size                    lhisize = 0;
+
+       if (pageop->btpo_flags & BTP_LEAF)
+       {
+               hsize += (sizeof(CommandId) + sizeof(RelFileNode));
+               memcpy(&hnode, (char*)xlrec + SizeOfBtreeSplit + 
+                                       sizeof(CommandId), sizeof(RelFileNode));
+       }
+       else
+       {
+               lhikey = (char*)xlrec + hsize;
+               memcpy(&btdata, lhikey, sizeof(BTItemData));
+               lhisize = IndexTupleDSize(btdata.bti_itup) +
+                                       (sizeof(BTItemData) - sizeof(IndexTupleData));
+               hsize += lhisize;
+       }
+
+       if (! P_RIGHTMOST(pageop))
+               PageIndexTupleDelete(page, P_HIKEY);
+
+       if (onleft)             /* skip target item */
+       {
+               memcpy(&btdata, (char*)xlrec + hsize, sizeof(BTItemData));
+               itemsz = IndexTupleDSize(btdata.bti_itup) +
+                                       (sizeof(BTItemData) - sizeof(IndexTupleData));
+               hsize += itemsz;
+       }
+
+       for (item = (char*)xlrec + hsize; ; )
+       {
+               memcpy(&btdata, item, sizeof(BTItemData));
+               for (offno = P_FIRSTDATAKEY(pageop);
+                        offno <= maxoff;
+                        offno = OffsetNumberNext(offno))
+               {
+                       ItemId  lp = PageGetItemId(page, offno);
+                       BTItem  btitem = (BTItem) PageGetItem(page, lp);
+
+                       if (BTItemSame(&btdata, btitem))
+                       {
+                               PageIndexTupleDelete(page, offno);
+                               break;
+                       }
+               }
+
+               itemsz = IndexTupleDSize(btdata.bti_itup) +
+                                       (sizeof(BTItemData) - sizeof(IndexTupleData));
+               itemsz = MAXALIGN(itemsz);
+
+               if (item + itemsz < (char*)record + record->xl_len)
+               {
+                       previtem = item;
+                       item += itemsz;
+               }
+               else
+                       break;
+       }
+
+       /* time to insert hi-key */
+       if (pageop->btpo_flags & BTP_LEAF)
+       {
+               lhikey = (P_RIGHTMOST(pageop)) ? item : previtem;
+               memcpy(&btdata, lhikey, sizeof(BTItemData));
+               lhisize = IndexTupleDSize(btdata.bti_itup) +
+                                       (sizeof(BTItemData) - sizeof(IndexTupleData));
+       }
+
+       if (! _bt_add_item(page, 
+                       P_HIKEY,
+                       lhikey,
+                       lhisize,
+                       &hnode))
+               elog(STOP, "btree_split_redo: failed to add hi key to left sibling");
+
+       return;
+}
+
 #endif

diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile
index 303936f4553e3bfdb3e76b378604f48d05133e95..272a883c521d43a23a21bb913e27e9db9951d9fe 100644 (file)
--- a/src/backend/access/transam/Makefile
+++ b/src/backend/access/transam/Makefile
@@ -4,7 +4,7 @@
 #    Makefile for access/transam
 #
 # IDENTIFICATION
-#    $Header: /cvsroot/pgsql/src/backend/access/transam/Makefile,v 1.12 2000/08/31 16:09:46 petere Exp $
+#    $Header: /cvsroot/pgsql/src/backend/access/transam/Makefile,v 1.13 2000/10/13 12:05:21 vadim Exp $
 #
 #-------------------------------------------------------------------------
 
@@ -12,7 +12,7 @@ subdir = src/backend/access/transam
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = transam.o transsup.o varsup.o xact.o xid.o xlog.o rmgr.o
+OBJS = transam.o transsup.o varsup.o xact.o xid.o xlog.o xlogutils.o rmgr.o
 
 all: SUBSYS.o
 

diff --git a/src/include/access/nbtree.h b/src/include/access/nbtree.h
index 4ca61e0c630c5f6c6d725f7652ffdcaf3ab4fa92..02a0bb6f5d8c3d4fe553c4b80d2546761d3f0b6a 100644 (file)
--- a/src/include/access/nbtree.h
+++ b/src/include/access/nbtree.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: nbtree.h,v 1.44 2000/10/13 02:03:02 vadim Exp $
+ * $Id: nbtree.h,v 1.45 2000/10/13 12:05:22 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -259,8 +259,9 @@ typedef struct xl_btree_insert
 
 /* 
  * This is what we need to know about insert with split - 
- * 22 + [4+8] + [btitem] + right sibling btitems. Note that we need in
- * CommandID and HeapNode (4 + 8 bytes) only for leaf page insert.
+ * 22 + {4 + 8 | left hi-key} + [btitem] + right sibling btitems. Note that
+ * we need in CommandID and HeapNode (4 + 8 bytes) for leaf pages
+ * and in left page hi-key for non-leaf ones.
  */
 typedef struct xl_btree_split
 {
@@ -271,8 +272,8 @@ typedef struct xl_btree_split
        /* 
         * We log all btitems from the right sibling. If new btitem goes on
         * the left sibling then we log it too and it will be the first
-        * BTItemData at the end of this struct, but after (for the leaf
-        * pages) CommandId and HeapNode.
+        * BTItemData at the end of this struct after CommandId and HeapNode
+        * on the leaf pages and left page hi-key on non-leaf ones.
         */
 } xl_btree_split;