Quadcap Embeddable Database

com/quadcap/sql/index/Bnode.java

Go to the documentation of this file.

package com.quadcap.sql.index;

/* Copyright 1997 - 2003 Quadcap Software.  All rights reserved.
 *
 * This software is distributed under the Quadcap Free Software License.
 * This software may be used or modified for any purpose, personal or
 * commercial.  Open Source redistributions are permitted.  Commercial
 * redistribution of larger works derived from, or works which bundle
 * this software requires a "Commercial Redistribution License"; see
 * http://www.quadcap.com/purchase.
 *
 * Redistributions qualify as "Open Source" under  one of the following terms:
 *   
 *    Redistributions are made at no charge beyond the reasonable cost of
 *    materials and delivery.
 *
 *    Redistributions are accompanied by a copy of the Source Code or by an
 *    irrevocable offer to provide a copy of the Source Code for up to three
 *    years at the cost of materials and delivery.  Such redistributions
 *    must allow further use, modification, and redistribution of the Source
 *    Code under substantially the same terms as this license.
 *
 * Redistributions of source code must retain the copyright notices as they
 * appear in each source code file, these license terms, and the
 * disclaimer/limitation of liability set forth as paragraph 6 below.
 *
 * Redistributions in binary form must reproduce this Copyright Notice,
 * these license terms, and the disclaimer/limitation of liability set
 * forth as paragraph 6 below, in the documentation and/or other materials
 * provided with the distribution.
 *
 * The Software is provided on an "AS IS" basis.  No warranty is
 * provided that the Software is free of defects, or fit for a
 * particular purpose.  
 *
 * Limitation of Liability. Quadcap Software shall not be liable
 * for any damages suffered by the Licensee or any third party resulting
 * from use of the Software.
 */

import java.io.IOException;
import java.io.PrintStream;

import com.quadcap.sql.file.Block;
import com.quadcap.sql.file.BlockFile;
import com.quadcap.sql.file.ByteUtil;
import com.quadcap.sql.file.SubPageManager;

import com.quadcap.util.Debug;
import com.quadcap.util.Util;

/**
 * This class represents a node in the btree.
 * <p>
 *
 * <b>Things to think about:</b>
 * <p>
 * <UL>
 * <LI>Key Compression using dictionary?  With a large enough node, say
 *      4 or 8 k, you could perform a block compression
 * <LI>Key Endianness (X.500 keys are LSB, most others are MSB)
 * <LI>Large data stored in overflow areas (blockstream)
 * <LI>Thread-safeness
 * <LI>Improve block garbage collection (it appears that some deleted
 *      keys aren't properly counted as garbage!)
 * </UL>
 *
 * @author Stan Bailes
 */

// struct node {
//     int      count;  /* number of items in this node */
//     int      dataBottom;     /* data area -- grows downward */
//     int       flags;         /* isRoot, */
//     int[]    keyIndex;       /* keyrefs, grows up */
//     byte[]   empty;          /* ... empty in the middle ... */
//     byte[]   data;           /* key + data, grows down */
// }

public class Bnode {
    //#ifdef DEBUG
    static final boolean paranoid = false;
    //#endif
    
    Btree                       tree;
    BlockFile                   file;
    long                        blockRef;

    /** The number of octets in the representation of a reference to
     * a byte offset in the data area of this block.   Currently,
     * bnode blocks can't be larger than 65536 because of the
     * two byte ref size.
     */
    public static final int     REF_SIZE                = 2;

    /** Number of key/data pairs in this node.  */
    static final int            fCount                  = 0;

    /** Bottom of data area -- grows downward. */
    static final int            fDataBOS                = 4;

    /** Flags */
    static final int            fFlags                  = 8;
    static final int                    IS_LEAF                 = 0x0001;
    static final int                    BNODE_MAGICF            = 0xff00;
    static final int                    BNODE_MAGICV            = 0xbd00;

    /**
     * Amount of space that could be reclaimed by repacking the
     * node.  When keys are deleted, we don't immediately remove
     * the key/data pair in the data area.  Instead, we remove the
     * pointer the pair from the key index list, and record (here)
     * the size of the key/data pair.  Later, when the block becomes
     * full, we reclaim this space by repacking the data area to
     * eliminate the holes.
     */
    static final int            fGarbage                = 12;

    static final int            fParent                 = 16;

    /** Start of key indices. */
    static final int            fIndices                = 24;
    

    /**
     * Construct a node from a buffer.
     * @param tree the Btree containing this node
     * @param blockRef the block number in <b>file</b> of this node.
     */
    public Bnode(Btree tree, long blockRef) {
        //#ifdef DEBUG
        if (Trace.bit(2)) {
            Debug.println("Bnode(" + blockRef + ")");
        }
        //#endif
        this.tree = tree;
        this.file = tree.file;
        this.blockRef = blockRef;       
    }

    /** <hr><font size =+1>Index Interface</font> <!----------------------!> */

    /**
     * Return the number of keys in this subtree.  This implicitly only
     * counts entries in the leaf nodes.
     */
    public int size() throws IOException {
        Block b = getBlock();
        try {
            return size(b);
        } finally {
            b.decrRefCount();
        }
    }
    
    public int size(Block b) throws IOException {
        int tot = 0;
        for (int i = 0; i < getCount(b); i++) {
            Block c = getBlock(blockRef(b, i));
            try {
                if (isLeaf(c)) {
                    tot += getCount(c);
                } else {
                    tot += size(c);
                }
            } finally {
                c.decrRefCount();
            }
        }
        return tot;
    }

    /**
     * Implement the get operation.
     * @param key the search key
     * @return if the get succeeds, the data is returned as a byte array;
     *          if the get fails, <b>null</b> is returned.
     */
    public final byte[] get(byte[] key, int klen) throws IOException {
        //#ifdef DEBUG
        if (Trace.bit(2)) {
            Debug.println("Bnode[" + blockRef + "] get(" +
                          Util.hexBytes(key) + ")");
        }
        //#endif
        Block b = getBlock();
        byte[] ret = null;
        try {
            ret = get(b, key, klen);
        } finally {
            b.decrRefCount();
        }
        return ret;
    }

//      /**
//       * Leaf level get operation
//       */
//      public final int get(byte[] key, int koff,
//                           byte[] data, int off, int len) throws IOException {
//      Block b = getBlock();
//          try {
//              return get(b, key, koff, data, off, len);
//          } finally {
//              b.decrRefCount();
//          }
//      }

    /**
     * Implement the set operation.
     * @param key the key
     * @param data the data
     */
    public final boolean set(byte[] key, int klen,
                             byte[] data, int doff, int dlen,
                             boolean insOk, boolean updOk) throws IOException {
        //#ifdef DEBUG
        if (Trace.bit(2)) {
            Debug.println("[" + blockRef + "] set(" +
                          Util.hexBytes(key) + ", " +
                          Util.hexBytes(data) + ")");
        }
        //#endif
        Block b = getBlock();
        try {
            return set(b, key, klen, data, doff, dlen, insOk, updOk);
        } finally {
            b.decrRefCount();
        }
    }

    /**
     * Implement the delete operation.
     * @param key the key
     */
    public final boolean delete(byte[] key) throws IOException {
        Block b = getBlock();
        try {
            return delete(b, key);
        } finally {
            b.decrRefCount();
        }
    }

    /**
     * A helper function to perform a single-level of the recursive search.
     * @param b the block to be searched.
     * @param key the search key
     * @return if found, the data, else null
     */
    final byte[] get(Block b, byte[] key, int klen) throws IOException {
        byte[] ret = null;
        int bs = bsearch(b, key, klen);
        if (!isLeaf(b)) {
            Block c = getSearchBlock(b, bs);
            try {
                ret = get(c, key, klen);
            } finally {
                c.decrRefCount();
            }
        } else if (bs >= 0) {
            ret = dataAtPos(b, bs);
        }
        //#ifdef DEBUG
        if (paranoid) { checkBlock(b); }
        //#endif
        return ret;
    }

    final Block getSearchBlock(Block b, int bs) throws IOException {
        int index = bs < 0 ? (-1-bs) : bs;
        if (bs < 0) {
            index = index - 1;
            if (index < 0) {
                index = 0;
            }
        }
        Block c = getBlock(blockRef(b, index));
        return c;
    }

//      final int get(Block b, byte[] key, int koff,
//                    byte[] data, int off, int len)
//          throws IOException
//      {
//          int ret = -1;
//      int bs = bsearch(b, tree.compare, key, 0, getCount(b)-1);
//          if (!isLeaf(b)) {
//          Block c = getSearchBlock(b, bs);
//          try {
//              ret = get(c, key, koff, data, off, len);
//          } finally {
//              c.decrRefCount();
//          }
//      } else {
//              if (bs >= 0) {
//                  ret = dataAtPos(b, bs, koff, data, off, len);
//              } else {
//                  ret = -1;
//              }
//          }
//      return ret;
//      }

    /**
     * Initialize an empty <b>BNode</b>.
     * <p><b>PRECONDITION:</b>Hold lock on <code>b</code>.
     * @param b the block to be initialized.
     */
    final void init(Block b, boolean isLeaf) throws IOException  {
        setCount(b, 0);
        setFlags(b, (isLeaf ? IS_LEAF : 0) | BNODE_MAGICV);
        setBos(b, file.getPageSize());
        setGarbage(b, 0);
        setParent(b, 0);
    }

    final void init(boolean f) throws IOException {
        Block b = getBlock();
        try {
            init(b, f);
        } finally {
            b.decrRefCount();
        }
    }

    final void checkMagic() throws IOException {
        Block b = getBlock();
        try {
            if ((getFlags(b) & BNODE_MAGICF) != BNODE_MAGICV) {
                throw new IOException("Not a valid index node: " +
                                      SubPageManager.toString(b.getPageNum()));
            }
        } finally {
            b.decrRefCount();
        }
    }
        
    /**
     * Free the resources associated with this block.  If the block is a
     * non-leaf node, free the entire subtree.
     */
    public final void free() throws IOException {
        free(blockRef);
    }
    
    /**
     * Free the resources associated with the specified block.  If the block is a
     * non-leaf node, free the entire subtree.
     */
    private final void free(long blockNum) throws IOException {
        //#ifdef DEBUG
        if (blockNum == 0) {
            Debug.println("************************** Trying to free block 0!");
            return;
        }
        //#endif
        Block b = getBlock(blockNum);
        //#ifdef DEBUG
        if (Trace.bit(2)) {
            Debug.println("free(" + blockNum + ")");
            display(b, Debug.debugStream, false);
        }
        //#endif
        try {
            if (!isLeaf(b)) {
                for (int i = 0; i < getCount(b); i++) {
                    free(blockRef(b, i));
                }
            }
            file.freePage(b.getPageNum());
        } finally {
            b.decrRefCount();
        }
    }

    /**
     * Perform a binary search of the keys in the specified block, searching
     * for the given key.  If the key is found in the block, return the
     * index of the matching key.  If the key isn't
     * found, return < 0, and for the key index which is the smallest
     * existing key greater than the given key, return
     * <code>0 - (index+1)</code>.
     * @param b the block to be searched.
     * @param key the search key
     * @param lo index of smallest element in search region
     * @param hi index of largest element in search region
     * @return if found, the index of the matching key/data pair. If not
     *          found, return the index where the data would be found
     *          as an expression of the form:<p>
     *          <code>0 - (index + 1)</code>
     */
    final static int bsearch(Block b, Comparator c, byte[] key,
                             int klen, int lo, int hi)
        throws IOException
    {
        while (hi >= lo) {
            int mid = (hi + lo) / 2;
            int ret = keyCompareAtPos(c, key, klen, b, mid);
            if (ret < 0) hi = mid-1;
            else if (ret > 0) lo = mid+1;
            else return mid;
        }
        return 0 - (hi+2);
    }

    /**
     * Perform a binary search of the keys in the specified block, searching
     * for the given key.  If the key is found in the block, return the
     * index of the matching key.  If the key isn't
     * found, return < 0, and for the key index which is the smallest
     * existing key greater than the given key, return
     * <code>0 - (index+1)</code>.
     * @param b the block to be searched.
     * @param key the search key
     * @return if found, the index of the matching key/data pair. If not
     *          found, return the index where the data would be found
     *          as an expression of the form:<p>
     *          <code>0 - (index + 1)</code>
     */
    final int bsearch(Block b, byte[] key, int klen) throws IOException {
        return bsearch(b, tree.compare, key, klen, 0, getCount(b)-1);
    }

    /**
     * Return the data for a specified key as an integer
     */
    final static long blockRef(Block b, int index) throws IOException {
        return longDataAtPos(b, index);
    }

    /**
     * Recursive implementation of <code>set(key, data)</code>.<p>
     * <b>Precondition</b>: The caller must already have a write-lock on
     *  <code>b</code>.
     * @param b the block to be searched
     * @param key the search key
     * @param data the data associated with the key
     * @return true if the key already existed before the set, false otherwise.
     */
    final boolean set(Block b, byte[] key, int klen,
                      byte[] data, int doff, int dlen,
                      boolean insOk, boolean updOk) throws IOException {
        //#ifdef DEBUG
        byte[] save = null;
        if (paranoid) {
            byte[] d = (byte[])b.getData();
            save = new byte[d.length];
            System.arraycopy(d, 0, save, 0, d.length);
        }
        //#endif
        int bs = bsearch(b, key, klen);
        if (bs < 0 && !insOk) {
            throw new IOException("Key not found");
        } else if (bs >= 0 && !updOk) {
            throw new IOException("Key not unique"); 
        }
        int index = bs < 0 ? (-1-bs) : bs;
        boolean ret = false;
        if (isLeaf(b)) {
            boolean replace = (bs >= 0);
            setKey(b, index, key, klen, data, doff, dlen, replace);
            ret = replace;
        } else {
            if (bs < 0) {
                index = index - 1;
                if (index < 0) index = 0;
            }
            Block c = getBlock(blockRef(b, index));
            try {
                ret = set(c, key, klen, data, doff, dlen, insOk, updOk);
            } finally {
                c.decrRefCount();
            }
        }
        //#ifdef DEBUG
        if (paranoid) {
            try {
                checkBlock(b);
            } catch (RuntimeException e) {
                b.setData(save);
                Debug.println("bs = " + bs);
                Debug.println("index = " + index);
                Debug.println("isLeaf() = " + isLeaf(b));
                Debug.println("key = " + Util.hexBytes(key));
                Debug.println("data = " + Util.hexBytes(data));
                Debug.println("checkSpace = " +
                              debugcheckSpace(b, index, klen, dlen, bs>=0));
                Debug.println("old block:");
                display(b, Debug.debugStream, false);
            }
        }
        //#endif
        return ret;
    }

    /**
     * Recursive implementation of <code>delete(key)</code>.
     * @param b the block to be searched
     * @param key the search key
     * @return true if the key already existed before the delete,
     *                  false otherwise.
     */
    final boolean delete(Block b, byte[] key) throws IOException {
        boolean ret = false;
        boolean del = false;
        int bs = bsearch(b, key, key.length);
        int index = bs < 0 ? (-1-bs) : bs;
//      Debug.println(2, "delete([" + b.getPageNum() + "] " +
//                    keybytes(key) +
//                    "): bs = " + bs + ", index = " + index);
        if (isLeaf(b)) {
            if (bs >= 0) {
                del = deleteKeyAtPos(b, index, false);
                ret = true;
            }
        } else {
            if (bs < 0) {
                index = index - 1;
                if (index < 0) index = 0;
            }
            Block c = getBlock(blockRef(b, index));
            try {
                ret = delete(c, key);
            } finally {
                c.decrRefCount();
            }
        }
        //#ifdef DEBUG
        if (paranoid && !del) checkBlock(b);
        //#endif
        return ret;
    }

    //#ifdef DEBUG
    final void checkBlock(Block b) throws IOException {
        int g1 = getGarbage(b);
        int tk = 0;
        for (int i = 0; i < getCount(b); i++) {
            tk += existingKeyLength(b, i);
            if (i > 0) {
                byte[] k1 = keyAtPos(b, i-1);
                byte[] k2 = keyAtPos(b, i);
                try {
                    if (tree.compare.compare(k1, 0, k1.length,
                                             k2, 0, k2.length) >= 0) {
                        try {
                            display(b, System.out, false);
                        } catch (Exception e) {}
                        throw new RuntimeException("Bad key ordering");
                    }
                } catch (RuntimeException e) {
                    try {
                        display(b, System.out, false);
                    } catch (Exception ee) {
                    }
                    throw e;
                }
            }
        }
        int sk = file.getPageSize() - getBos(b);
        if (tk + g1 != sk) {
            try {
                display(b, System.out, false);
            } catch (Exception e) {}
            throw new RuntimeException("checkBlock: failure, " + (tk+g1) + " != " +
                                       sk);
        }

    }
    //#endif

    /**
     * Given a block and a key index, return the data for that index as a new
     * byte array.
     * @param b the block on which to operate.
     * @param index the index of the item for which the data is to be retrieved
     * @return the data for the key at the specified position in the node.
     */
    final static byte[] dataAtPos(Block b, int index) {
        int keyIndex = getKeyIndex(b, index);

        int[] start = new int[1];
        int keylen = getKeyLen(b, keyIndex, start);
        int keystart = start[0];

        int datastart = keystart + keylen;
        int datalen = getKeyLen(b, datastart, start);
        datastart = start[0];
        byte[] data = new byte[datalen];
        b.read(datastart, data, 0, data.length);
        return data;
    }

    final static int dataAtPos(Block b, int index, int koff,
                               byte[] data, int off, int len)
    {
        int keyIndex = getKeyIndex(b, index);

        int[] start = new int[1];
        int keylen = getKeyLen(b, keyIndex, start);
        int keystart = start[0];

        int datastart = keystart + keylen;
        datastart = start[0];

        return b.read(datastart + koff, data, off, len);
    }

    /**
     * Given a block and a key index, return the data for that index as
     * a long.
     * @param b the block on which to operate.
     * @param index the index of the item for which the data is to be retrieved
     * @return the data for the key at the specified position in the node.
     */
    final static long longDataAtPos(Block b, int index) {
        int keyIndex = getKeyIndex(b, index);

        keyIndex = getKeyEnd(b, keyIndex);      // skip to data
        if (b.readByte(keyIndex) != 8) {
            throw new RuntimeException("not a long: " + keyIndex + ": " + b.readByte(keyIndex));
        }
        return b.readLong(keyIndex + 1);
    }

    /**
     * Compute the total length of an existing key data pair, including
     * both key and data length fields.
     * @param b the block on which to operate.
     * @param index the index of the item for which the data is to be retrieved
     * @return the total number of data area bytes consumed by this item.
     */
    final static int existingKeyLength(Block b, int index) {
        int keyIndex = getKeyIndex(b, index);

        int[] start = new int[1];
        int keylen = getKeyLen(b, keyIndex, start);
        int keystart = start[0];

        int datastart = keystart + keylen;
        int datalen = getKeyLen(b, datastart, start);
        datastart = start[0];
        int end = datastart + datalen;
        return end - keyIndex;
    }

    /**
     * Given a block and a key index, return the key for that index as a new
     * byte array.
     * @param b the block on which to operate.
     * @param index the index of the item for which the key is to be retrieved
     */
    final static byte[] keyAtPos(Block b, int index) {
        int keyIndex = getKeyIndex(b, index);

        int[] start = new int[1];
        int keylen = getKeyLen(b, keyIndex, start);
        int keystart = start[0];

        byte[] key = new byte[keylen];
        b.read(keystart, key, 0, key.length);
        return key;
    }

    final static int getBytes(Block b, int pos, byte[] buf, int[] lenret) {
        int len = b.readByte(pos++) & 0xff;
        if ((len & 0x80) != 0) {
            int cnt = len & 0x7f;
            len = 0;
            while (cnt-- > 0) {
                len <<= 8;
                len += (b.readByte(pos++) & 0xff);
            }
        }
        lenret[0] = len;
        if (len > buf.length) return 0 - len;
        b.read(pos, buf, 0, len);
        return pos + len;
    }
        
    /**
     * Given a block and a key index, return the key for that index as a new
     * byte array.
     * @param b the block on which to operate.
     * @param index the index of the item for which the key is to be retrieved
     * @param buf the buffer into which the key bytes should be places
     * @param off the offset into the buffer
     * @param len the maximum number of bytes to write to the buffer
     *
     * @return the number of bytes returned, if sucessful.  If the buffer
     *    isn't big enough, we return a negative number '0 - k', where
     *    'k' is the actual key length.
     */
    public final static boolean getKeyAndData(Block b, int index,
                                              byte[] k, byte[] d,
                                              int[] lengths) {
        int pos = getKeyIndex(b, index);
        pos = getBytes(b, pos, k, lengths);
        int savelen = lengths[0];
        if (pos < 0) return false;
        pos = getBytes(b, pos, d, lengths);
        lengths[1] = lengths[0];
        lengths[0] = savelen;
        if (pos < 0) return false;
        return true;
    }

    /**
     * Perform a key comparison with the specified key in the block.
     * @param c the compare function
     * @param key the compare key
     * @param b the block to be compared
     * @param index the position of the key to be compared to the compare key
     *
     * @return the integer compare value (usu: -1, 0, 1)
     */
    static final int keyCompareAtPos(Comparator c, byte[] key, int klen,
                                     Block b, int index) throws IOException {
        int keypos = getKeyIndex(b, index);
        int r = getKeyLen(b, keypos);
        int start = (r >> 16) & 0xffff;
        int len = r & 0xffff;
        byte[] buf = (byte[])b.getData();
        int ret = c.compare(key, 0, klen, buf, start, len);
        return ret;
    }

    /**
     * Find the length of the specified key, as well as the byte offset
     * to the start of the key.
     *
     * @param b the B-node
     * @param keypos the buffer offset where the key/data pair starts.
     * @param keystart an <b>out</b> parameter, used to return the buffer
     *          offset where the actual key begins.
     */
    static final int getKeyLen(Block b, int keypos, int[] keystart) {
        int len = b.readByte(keypos++) & 0xff;
        if ((len & 0x80) != 0) {
            int cnt = len & 0x7f;
            len = 0;
            while (cnt-- > 0) {
                len <<= 8; len += (b.readByte(keypos++) & 0xff);
            }
        }
        keystart[0] = keypos;
        return len;
    }

    /**
     * Find the length of the specified key, as well as the byte offset
     * to the start of the key.
     *
     * @param b the B-node
     * @param keypos the buffer offset where the key/data pair starts.
     * @return the key length
     */
    final static int getKeyLen(Block b, int keypos) {
        int len = b.readByte(keypos++) & 0xff;
        if ((len & 0x80) != 0) {
            int cnt = len & 0x7f;
            len = 0;
            while (cnt-- > 0) {
                len <<= 8; len += (b.readByte(keypos++) & 0xff);
            }
        }
        return ((keypos & 0xffff) << 16) | (len & 0xffff);
    }

    /**
     * Find the length of the specified key, as well as the byte offset
     * to the start of the key.
     *
     * @param b the B-node
     * @param keypos the buffer offset where the key/data pair starts.
     * @return two 16-bit integers {start, len} encoded in a 32 bit int.
     */
    static final int getKeyEnd(Block b, int keypos) {
        int len = b.readByte(keypos++) & 0xff;
        if ((len & 0x80) != 0) {
            int cnt = len & 0x7f;
            len = 0;
            while (cnt-- > 0) {
                len <<= 8; len += (b.readByte(keypos++) & 0xff);
            }
        }
        return keypos + len;
    }

    /**
     * Compute the total number of bytes that will be required to store
     * the byte array 'b' plus its length code.
     * @param b the byte array
     * @return the length of the byte array plus the length of the length
     *          code.
     */
    final static int totalLength(int len) {
        int lenlen = 1;
        if (len > 0x7f) {
            lenlen++;
            if (len > 0xff) {
                lenlen++;
                if (len > 0xffff) {
                    lenlen++;
                    if (len > 0xffffff) {
                        lenlen++;
                    }
                }
            }
        }
        return len + lenlen;
    }

    /**
     * Insert a key/data pair at the bottom of the data area and
     * return its offset.  This method does <b>NOT</b> check the
     * capacity of the buffer -- it is assumed that the caller has
     * already ensured that enough space is available.
     * @param b the block to operate on 
     * @param key the search key
     * @param data the data associated with the key
     */
    final static int insertKeyData(Block b, byte[] key, int klen,
                                   byte[] data, int doff, int dlen) {
        // ---- insert the key/data pair at the bottom of the data area
        int bos = getBos(b) - dlen;
        b.write(bos, data, doff, dlen);
        bos = writeLenLen(b, bos, dlen) - klen;
        b.write(bos, key, 0, klen);
        bos = writeLenLen(b, bos, klen);
        setBos(b, bos);         // record the new data BOS.
        return bos;
    }

    /**
     * Insert or replace a key in the specified block, which may or may
     * not be a leaf block.  Handle split operations: if a split is required,
     * determine which post-split block this key should be added to.
     * Also, if the new key happens to be key zero (i.e., the lowest valued
     * key in this block), then adjust the parent's reference to this block
     * accordingly.  Although, maybe that can't happen?
     *
     * <p><b>PRECONDITION:</b> Lock/ref on block <code>b</code>
     * <p><b>POSTCONDITION:</b> Lock/ref on block <code>b</code>
     *
     * @return the block into which the key was actually stored.  This
     *          will be different from <code>b</code> if a split was
     *          necessary.
     */
    final Block setKey(Block b, int index, byte[] key, int klen,
                       byte[] data, int doff, int dlen,
                       boolean replace)
        throws IOException
    {
        Block r = b;
        if (index < 0) {
            index = bsearch(b, key, klen);
            if (index < 0) {
                replace = false;
                index = -1 - index;
            }
        }
        // if it looks like we're inserting a key lower than the current
        // low key, remember the old low key, we'll have to update the parent
        // node.
        byte[] okey = (index == 0 && getCount(b) > 0) ? keyAtPos(b, 0) : null;
            
        if (!setKeyAtPos(b, index, key, klen, data, doff, dlen, replace)) {
            if (replace) {
                if (deleteKeyAtPos(b, index, true)) {
                    //#ifdef DEBUG
                    Debug.println("Just lost b!");
                    //#endif
                }
            }
            Block[] ba = new Block[2];
            ba[0] = b;
            split(ba);
            b = ba[0];
            Block nb = ba[1];
            
            try {
                // After split, the key we were trying to add goes
                // into one of the two split blocks.
                int cv = keyCompareAtPos(tree.compare, key, klen, b, 0);
                r = cv < 0 ? nb : b;

                try {
                    int ret = bsearch(r, key, klen);
                    if (ret >= 0) {
                        //#ifdef DEBUG
                        Debug.println("After a split operation, the key was " +
                                      "found to be already present in the " +
                                      "target block at position " + ret);
                        Debug.println("The key: " + Util.hexBytes(key));
                        Debug.println("cv = " + cv);
                        Debug.println("split L: ");
                        display(ba[0], Debug.debugStream, false);
                        Debug.println("split R: ");
                        display(ba[1], Debug.debugStream, false);
                        //#endif
                        throw new RuntimeException(
                            "internal error in setKey: dup, key = " +
                            Util.strBytes(key));
                    }
                    ret = -1 - ret;
                    if (!setKeyAtPos(r, ret, key, klen, data, doff, dlen, false)) {
                        throw new RuntimeException(
                            "com.quadcap.sql.index.Bnode: " +
                            "internal error in setKey: no room.  " +
                            "key length = " + key.length +
                            ", data.length = " + data.length);
                    }
                    if (ret == 0 && getCount(r) > 1) {
                        newLow(r, keyAtPos(r, 1));
                    }
                } finally {
                }
            } finally {
                nb.decrRefCount();
            }
        } else {
            if (okey != null && index == 0) {
                /* && (!replace || getCount(b) > 1)) */
                newLow(b, okey);
            }
        }
        return r;
    }

    /**
     * Split this block, by creaing a new block and moving half of this
     * block's keys into the new block.  Then, add a new key to the parent
     * block to reflect the new block, and adjust the parent's old reference
     * to this block to reflect the new key distribution.
     *
     * <p><b>PRECONDITION:</b> We have a ref/lock on <code>ba[0]</code>
     * <p><b>POSTCONDITION:</b> We have a ref on <code>ba[1]</code>
     *
     * @param ba a two-element array, with <code>ba[0]</code> being
     *          the pre-split block, and <code>ba[1]</code> the new block
     */
    final void split(Block[] ba) throws IOException {
        Block b = ba[0];
        long nbno = file.newPage();
        //#ifdef DEBUG
        if (Trace.bit(5)) {
            Debug.println(0, "split: new block = " + nbno);
        }
        //#endif
        Bnode node = this.tree.getNode(nbno);
        node.init(isLeaf(b));
        Block nb = null;
        boolean gotException = true;
        try {
            nb = node.getBlock();       
            ba[1] = nb;
            splitHelp(ba, nbno);
            gotException = false;
        } finally {
            if (gotException) {
                if (nb != null) {
                    nb.decrRefCount();
                    ba[1] = null;
                }
                file.freePage(nbno);
            }
        }
    }
        
    /**
     * Split helper:
     * <p><b>PRECONDITION</b>: Ref/lock on both <code>ba[0], ba[1]</code>
     */
    final void splitHelp(Block[] ba, long nbno) throws IOException {
        Block b = ba[0];
        Block nb = ba[1];
        setParent(nb, getParent(b));

        //#ifdef DEBUG
        if (Trace.bit(5)) {
            Debug.println("BEFORE SPLIT: b");
            display(b, Debug.debugStream, false);
            Debug.println("BEFORE SPLIT: nb");
            display(nb, Debug.debugStream, false);
        }
        //#endif
        
        int more = capacity(b);
        int ncnt = 0;
        for (int i = 0; capacity(nb) > more + getGarbage(b); i++) {
            byte[] key = keyAtPos(b, i);
            byte[] data = dataAtPos(b, i);
            setKeyAtPos(nb, i, key, key.length, data, 0, data.length, false);
            if (!isLeaf(nb)) {
                Bnode child = this.tree.getNode(Util.integer(data));
                Block cb = child.getBlock();
                try {
                    Bnode.setParent(cb, nbno);
                } finally {
                    cb.decrRefCount();
                }
            }
            forgetKeyAtPos(b, i);
            more += REF_SIZE;
            ncnt++;
        }
        int cnt = getCount(b);
        int len = cnt - ncnt;
        moveKeys(b, ncnt, 0, len);
        setCount(b, len);
        gc(b);
        //#ifdef DEBUG
        if (paranoid) try { 
            checkBlock(b); 
        } catch (RuntimeException e) {
            Debug.print(e);
        }
        //#endif

        propogateSplit(ba);
    }

    /**
     * After the split operation, propagate information up the tree to
     * the parent block about the newly created block and the new key
     * distribution.
     *
     * @param ba a two-element array, with <code>ba[0]</code> being
     *          the pre-split block, and <code>ba[1]</code> the new block
     */
    final void propogateSplit(Block[] ba) throws IOException {
        Block b = ba[0];
        Block nb = ba[1];
        long b_blk = b.getPageNum();
        long parentBlock = getParent(b);
        Bnode pnode = null;
        if (parentBlock == 0) {
            // ---- We're splitting the root block.
            // ---- We'd like for block 'b', which is currently the parent,
            // ---- to remain so.  So we'll move the data in 'b' to 'b2',
            // ---- then rename 'b' to 'parent', and 'b2' to 'b'.
            // ---- I think.
            long b2 = file.newPage();
            Block b2b = getBlock(b2);

            // ---- no need to synchronize here, since we already hold the
            // ---- lock on the parent node, and nobody else knows about
            // ---- b2 yet.
            try {
                b2b.takeData(b);
                parentBlock = b.getPageNum();
                setParent(b2b, parentBlock);
                pnode = this.tree.getNode(parentBlock);
                pnode.init(false);
                ba[0] = b = b2b;
                if (!isLeaf(b2b)) for (int i = 0; i < getCount(b2b); i++) {
                    long cref = blockRef(b2b, i);
                    Block c = getBlock(cref);
                    try {
                        setParent(c, b2);
                    } finally {
                        c.decrRefCount();
                    }
                }
                b_blk = b2;
            } finally {
                b2b.decrRefCount();
            }
                    
        } else {
            pnode = this.tree.getNode(parentBlock);
        }

        Block pb = pnode.getBlock();
        try {
            // ---- the parent currently has an entry for the old block,
            // ---- but with the key that is now associated with the new
            // ---- block.  So we replace the old key, with the new block
            // ---- value, and we add a new key, with the old block value.

            // ---- replace the old key
            byte[] okey = keyAtPos(nb, 0);
            byte[] odat = Util.bytes(nb.getPageNum());
            Block r = setKey(pb, -1, okey, okey.length, odat, 0, odat.length,
                             true);
            try {
                setParent(nb, r.getPageNum());

                // ---- add the new key
                byte[] nkey = keyAtPos(b, 0);
                byte[] ndat = Util.bytes(b.getPageNum());

                if (r != pb) {
                    // if the parent block was split as a result of
                    // the previous setKey, then it is possible that
                    // 'b' and 'nb' were on the boundary of that
                    // split, and will now have different parent
                    // blocks.  So we need to figure out which parent
                    // 'b' should get...

                    throw new RuntimeException("internal error");
                }
                r = setKey(r, -1, nkey, nkey.length, ndat, 0, ndat.length,
                           false);
                setParent(b, r.getPageNum());
            } finally {
            }
            
        } finally {
            pb.decrRefCount();
        }

    }

    /**
     * Reclaim the data space occupied by deleted keys.
     */
    final void gc(Block b) throws IOException {
        //#ifdef DEBUG
        byte[] saved = null;
        if (paranoid) {
            saved = new byte[file.getPageSize()];
            System.arraycopy((byte[])b.getData(), 0, saved, 0, saved.length);
        }
        //#endif

        int initcap = capacity(b);
        int amt = getGarbage(b);
        int cnt = getCount(b);
        if (amt > 0) {
            int size = 0;
            Object[][] pairs = new Object[2][cnt];
            for (int i = 0; i < cnt; i++) {
                size += existingKeyLength(b, i);
                pairs[0][i] = keyAtPos(b, i);
                pairs[1][i] = dataAtPos(b, i);
            }
            setBos(b, file.getPageSize());
            for (int i = 0; i < cnt; i++) {
                byte[] key = (byte[])pairs[0][i];
                byte[] data = (byte[])pairs[1][i];
                int pos = insertKeyData(b, key, key.length,
                                        data, 0, data.length);
                setKeyIndex(b, i, pos);
            }
        }
        setGarbage(b, 0);
    }

    /**
     * When an operation results in a change in the smallest key in a block,
     * it is necessary to inform the parent block of the change, since the
     * parent block key for this block is required to be the key of the
     * smallest item in this block.
     *
     * @param b the block containing the new lowest key
     * @param prevLow the key which was previously the lowest key.
     *
     * @return true if block <code>b</code> is now empty, and has been
     *          freed and unlinked from the tree.  Don't touch this block
     *          any more!       
     */
    final boolean newLow(Block b, byte[] prevLow) throws IOException {
        boolean ret = false;
        long parentBlock = getParent(b);
        if (parentBlock != 0) {
            Bnode parent = tree.getNode(parentBlock);
            Block pb = parent.getBlock();
            try {
                int pos = bsearch(pb, prevLow, prevLow.length);
                if (pos < 0) {
                    //#ifdef DEBUG
                    System.out.print("b = ");
                    display(b, System.out, false);
                    System.out.print("pb = ");
                    display(pb, System.out, false);
                    Debug.println(0, "prevLow = " + keybytes(prevLow));
                    //#endif
                    throw new RuntimeException(
                        "deleteKeyAtPos: deleting " +
                        " previous low key, not found!");
                }
                if (getCount(b) == 0) {
                    if (deleteKeyAtPos(pb, pos, false)) {
                        //#ifdef DEBUG
                        //Debug.println("Just deleted pb!");
                        //#endif
                    }
                    file.freePage(b.getPageNum());
                    ret = true;
                } else {
                    byte[] k = keyAtPos(b, 0);
                    byte[] d = Util.bytes(b.getPageNum());
                    setKey(pb, pos, k, k.length, d, 0, d.length, true);
                    //#ifdef DEBUG
                    if (paranoid) checkBlock(b);
                    //#endif
                }
            } finally {
                pb.decrRefCount();
            }
        }
        return ret;
    }

    /**
     * Helper function to move key indices around when adding or deleting
     * keys.
     */
    final static void moveKeys(Block b, int from, int to, int length) {
        if (length > 0) {
            byte[] buf = (byte[])b.getData();
            length *= REF_SIZE;
            int f = index(from);
            int t = index(to);
            if (f < t) {
                b.setDirty(true);
                for (int i = length-1; i >= 0; i--) {
                    buf[t+i] = buf[f+i];
                }
            } else if (f > t) {
                b.setDirty(true);
                for (int i = 0; i < length; i++) {
                    buf[t+i] = buf[f+i];
                }
            }
        }
    }

    /**
     * Determine if there is room in the block for a new key/data pair.
     * @param b the block to operate on
     * @param index the index of the key that is being replaced,
     *          if <code>replace</code> is <b>true</b>.
     * @param klen the search key length
     * @param dlen the length of the data associated with the key
     * @param replace <b>true</b> if an existing key/data pair is to
     *          be replaced by the new key, <b>false</b> otherwise.
     *
     * @return negative number if there is no room for the new data.<p>
     *          zero if there is room for the new data.<p>
     *         positive number if there would be room after a garbage
     *          collection.  Included in this calculation is the deletion
     *          of the key being replaced, if <code>replace</code> is true.<p>
     */
    final int checkSpace(Block b, int index, int klen, int dlen, boolean replace) {
        int need = REF_SIZE + totalLength(klen) + totalLength(dlen);
        int total = capacity(b) + getGarbage(b);
        if (replace) total += REF_SIZE + existingKeyLength(b, index);
        int ret = 0;                    // fits
        if (need > total) {
            ret = -1;                   // doesn't fit
        } else if (need > capacity(b)) {
            ret = 1;                    // will fit if gc
        }
        return ret;
    }

    //#ifdef DEBUG
    final int debugcheckSpace(Block b, int index, int klen, int dlen,
                       boolean replace) {
        int need = REF_SIZE + totalLength(klen) + totalLength(dlen);
        Debug.println(0, "keylen = " + totalLength(klen));
        Debug.println(0, "datalen = " + totalLength(dlen));
        Debug.println(0, "need = " + need);
        int total = capacity(b) + getGarbage(b);
        Debug.println(0, "capacity = " + capacity(b));
        Debug.println(0, "garbage = " + getGarbage(b));
        Debug.println(0, "total = " + total);
        if (replace) total += REF_SIZE + existingKeyLength(b, index);
        int ret = 0;                    // fits
        if (need > total) {
            ret = -1;                   // doesn't fit
        } else if (need > capacity(b)) {
            ret = 1;                    // will fit if gc
        }
        return ret;
    }
    //#endif

    /**
     * This function is scary.  It accounts for the fact that we're about
     * to delete a key, by including the size of the key/data in the
     * node's <code>garbage</code> field, but it doesn't actually delete
     * the key.  Be careful.
     */
    final void forgetKeyAtPos(Block b, int index) {
        setGarbage(b, getGarbage(b) + existingKeyLength(b, index));
    }

    /**
     * Delete the specified key.
     *
     * @return true if the block is now empty, has been freed, and we
     *          should avoid touching it...
     */
    final boolean deleteKeyAtPos(Block b, int index, boolean skipNewLow)
        throws IOException
    {
        boolean ret = false;
        int cnt = getCount(b);
        byte[] prevLow = index == 0 ? keyAtPos(b, 0) : null;
        forgetKeyAtPos(b, index);
        if (index < cnt-1) {
            moveKeys(b, index+1, index, cnt-index-1);
        }
        setCount(b, cnt-1);
        if (!skipNewLow && getCount(b) == 0 && getParent(b) == 0) {
            setLeaf(b, true);
        } else if (!skipNewLow && index == 0) {
            ret = newLow(b, prevLow);
        }
        return ret;
    }

    /**
     * Given a block and a key position, insert a new key/data pair
     * immediately at the specified position, returning true if the
     * key/data fit in the block.  If the incoming data won't fit,
     * don't modify the block and return false.