package com.dbxml.db.common.btree;

/*
 * dbXML - Native XML Database
 * Copyright (C) 1999-2004  The dbXML Group, L.L.C.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 *
 * $Id: BTree.java,v 1.4 2004/07/20 20:19:33 bradford Exp $
 */

import java.io.*;

import com.dbxml.db.core.DBException;
import com.dbxml.db.core.FaultCodes;
import com.dbxml.db.core.data.Value;
import com.dbxml.db.core.filer.FilerException;
import com.dbxml.db.core.indexer.IndexQuery;
import com.dbxml.db.core.indexer.IndexValue;
import com.dbxml.db.core.transaction.Transaction;
import com.dbxml.util.Array;
import com.dbxml.util.SoftHashMap;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Map;

/**
 * BTree represents a Variable Magnitude Simple-Prefix B+Tree File.  A
 * BTree is a bit flexible in that it can be used for set or map-based
 * indexing.  The Indexers use BTree as a map for indexing entity and
 * attribute values in Documents.
 * <br><br>
 * For those who don't know how a Simple-Prefix B+Tree works, the primary
 * distinction is that instead of promoting actual keys to branch pages,
 * when leaves are split, a shortest-possible separator is generated at
 * the pivot.  That separator is what is promoted to the parent branch
 * (and continuing up the list).  As a result, actual keys and pointers
 * can only be found at the leaf level.  This also affords the index the
 * ability to ignore costly merging and redistribution of pages when
 * deletions occur.  Deletions only affect leaf pages in this
 * implementation, and so it is entirely possible for a leaf page to be
 * completely empty after all of its keys have been removed.
 * <br><br>
 * Also, the Variable Magnitude attribute means that the btree attempts
 * to store as many values and pointers on one page as possible.  When
 * storing values, the largest common prefix between the values is
 * stored only once at the beginning of the page, with the remaining
 * value content being stored in the individual entries.  This method is
 * highly efficient for tight sequences and highly redundant values.
 */

public class BTree extends Paged {
   private static final byte[] EmptyBytes = new byte[0];
   private static final Value EmptyValue = new Value(EmptyBytes);

   protected static final byte LEAF = 1;
   protected static final byte BRANCH = 2;
   protected static final byte STREAM = 3;

   private Map cache = new SoftHashMap();
   private boolean compoundVals;

   private BTreeFileHeader fileHeader;
   private BTreeRootInfo rootInfo;
   private BTreeNode rootNode;

   public BTree(boolean compoundKeys) {
      super();
      this.compoundVals = compoundKeys;
      fileHeader = (BTreeFileHeader)getFileHeader();
      fileHeader.setPageCount(1);
      fileHeader.setTotalCount(1);
   }

   public BTree(File file, boolean compoundKeys) {
      this(compoundKeys);
      setFile(file);
   }

   public boolean open() throws DBException {
      if ( super.open() ) {
         Transaction tx = new Transaction();
         try {
            long p = fileHeader.getRootPage();
            rootInfo = new BTreeRootInfo(p);
            rootNode = getBTreeNode(tx, rootInfo, p, null);
            tx.commit();
            return true;
         }
         catch ( DBException e ) {
            tx.cancel();
            throw e;
         }
      }
      else
         return false;
   }

   public boolean create() throws DBException {
      if ( super.create() ) {
         Transaction tx = new Transaction();
         try {
            open();
            long p = fileHeader.getRootPage();
            rootInfo = new BTreeRootInfo(p);
            rootNode = new BTreeNode(rootInfo, getPage(tx, p));
            rootNode.ph.setStatus(LEAF);
            rootNode.setValues(new Value[0]);
            rootNode.setPointers(new long[0]);
            rootNode.write(tx);
            tx.commit();
            close();
            return true;
         }
         catch ( DBException e ) {
            tx.cancel();
            throw e;
         }
         catch ( Exception e ) {
            tx.cancel();
            e.printStackTrace(System.err);
            throw new FilerException(FaultCodes.COL_CANNOT_CREATE, e);
         }
      }
      return false;
   }

   /**
    * addValue adds a Value to the BTree and associates a pointer with
    * it.  The pointer can be used for referencing any type of data, it
    * just so happens that dbXML uses it for referencing pages of
    * associated data in the BTree file or other files.
    *
    * @param value The Value to add
    * @param pointer The pointer to associate with it
    * @return The previous value for the pointer (or -1)
    */
   public final long addValue(Transaction tx, Value value, long pointer) throws DBException, IOException {
      return getRootNode().addValue(tx, value, pointer);
   }

   /**
    * removeValue removes a Value from the BTree and returns the
    * associated pointer for it.
    *
    * @param value The Value to remove
    * @return The pointer that was associated with it
    */
   public final long removeValue(Transaction tx, Value value) throws DBException, IOException {
      return getRootNode().removeValue(tx, value);
   }

   /**
    * findValue finds a Value in the BTree and returns the associated
    * pointer for it.
    *
    * @param value The Value to find
    * @return The pointer that was associated with it
    */
   public final long findValue(Transaction tx, Value value) throws IOException, BTreeException {
      return getRootNode().findValue(tx, value);
   }

   /**
    * query performs a query against the BTree and performs callback
    * operations to report the search results.
    *
    * @param query The IndexQuery to use (or null for everything)
    * @param callback The callback instance
    */
   public final void query(Transaction tx, IndexQuery query, BTreeCallback callback) throws IOException, BTreeException {
      if ( compoundVals ) {
         int op = query.getOperator();
         if ( op == IndexQuery.EQ )
            query = new CompoundEQ(query.getPattern(), query.getValue(0));
         else if ( op == IndexQuery.NEQ )
            query = new CompoundNEQ(query.getPattern(), query.getValue(0));
      }
      getRootNode().query(tx, query, callback);
   }

   /**
    * setRootNode resets the root for the specified root object to the
    * provided BTreeNode's page number.
    *
    * @param root The root to reset
    * @param rootNode the new root node to use
    */
   protected final void setRootNode(Transaction tx, BTreeNode newRoot) throws DBException, IOException {
      BTreeRootInfo parent = rootInfo.getParent();
      if ( parent == null ) {
         rootNode = newRoot;
         long p = rootNode.page.getPageNum();
         rootInfo.setPage(p);
         fileHeader.setRootPage(p);
      }
      else {
         long p = newRoot.page.getPageNum();
         rootInfo.setPage(p);
         addValue(tx, rootInfo.name, p);
      }
   }

   /**
    * getRootNode retreives the BTree node for the file's root.
    *
    * @return The root node
    */
   protected final BTreeNode getRootNode() {
      return rootNode;
   }

   private BTreeNode getBTreeNode(Transaction tx, BTreeRootInfo root, long page, BTreeNode parent) {
      try {
         BTreeNode node;
         synchronized ( this ) {
            Long pNum = new Long(page);
            node = (BTreeNode)cache.get(pNum);
            if ( node == null ) {
               Page p = getPage(tx, pNum);
               node = new BTreeNode(root, p, parent);
            }
            else {
               node.root = root;
               node.parent = parent;
            }
         }
         synchronized ( node ) {
            if ( !node.isLoaded() ) {
               node.read(tx);
               node.setLoaded(true);
            }
         }
         return node;
      }
      catch ( Exception e ) {
         e.printStackTrace(System.err);
         return null;
      }
   }

   private BTreeNode createBTreeNode(Transaction tx, BTreeRootInfo root, byte status, BTreeNode parent) {
      try {
         Page p = getFreePage(tx);
         BTreeNode node = new BTreeNode(root, p, parent);
         node.ph.setStatus(status);
         node.setValues(new Value[0]);
         node.setPointers(new long[0]);
         return node;
      }
      catch ( Exception e ) {
         e.printStackTrace(System.err);
         return null;
      }
   }


   /**
    * BTreeRootInfo
    */

   private class BTreeRootInfo {
      private BTreeRootInfo parent;
      private Value name;
      private long page;

      public BTreeRootInfo(BTreeRootInfo parent, String name, long page) {
         this.parent = parent;
         this.name = new Value(name);
         this.page = page;
      }

      public BTreeRootInfo(BTreeRootInfo parent, Value name, long page) {
         this.parent = parent;
         this.name = name;
         this.page = page;
      }

      public BTreeRootInfo(String name, long page) {
         this.parent = rootInfo;
         this.name = new Value(name);
         this.page = page;
      }

      public BTreeRootInfo(Value name, long page) {
         this.parent = rootInfo;
         this.name = name;
         this.page = page;
      }

      private BTreeRootInfo(long page) {
         parent = null;
         name = null;
         this.page = page;
      }

      public synchronized BTreeRootInfo getParent() {
         return parent;
      }

      public synchronized Value getName() {
         return name;
      }

      public synchronized long getPage() {
         return page;
      }

      public synchronized void setPage(long page) {
         this.page = page;
      }
   }


   /**
    * BTreeNode
    */

   private class BTreeNode {
      private BTreeRootInfo root;
      private Page page;
      private BTreePageHeader ph;
      private Value[] values;
      private Value prefix;
      private long[] ptrs;
      private BTreeNode parent;
      private boolean loaded;

      public BTreeNode(BTreeRootInfo root, Page page, BTreeNode parent) {
         this.root = root;
         this.page = page;
         this.parent = parent;
         ph = (BTreePageHeader)page.getPageHeader();
      }

      public BTreeNode(BTreeRootInfo root, Page page) {
         this.root = root;
         this.page = page;
         ph = (BTreePageHeader)page.getPageHeader();
      }

      private synchronized void setValues(Value[] values) {
         this.values = values;
         ph.setValueCount((short)values.length);
         if ( values.length > 1 )
            prefix = getPrefix(values[0], values[values.length-1]);
         else
            prefix = EmptyValue;
         ph.setPrefixLength((short)prefix.getLength());
      }

      private synchronized void setPointers(long[] ptrs) {
         this.ptrs = ptrs;
      }

      private synchronized boolean isLoaded() {
         return loaded;
      }

      private synchronized void setLoaded(boolean loaded) {
         this.loaded = loaded;
      }

      public synchronized void read(Transaction tx) throws DBException, IOException {
         Value v = readValue(tx, page);
         DataInputStream is = new DataInputStream(v.getInputStream());

         // Read in the common prefix (if any)
         short pfxLen = ph.getPrefixLength();
         byte[] pfxBytes;
         if ( pfxLen > 0 ) {
            pfxBytes = new byte[pfxLen];
            is.read(pfxBytes);
         }
         else {
            prefix = EmptyValue;
            pfxBytes = EmptyBytes;
         }

         // Read in the pointers
         ptrs = new long[ph.getPointerCount()];
         for ( int i = 0; i < ptrs.length; i++ )
            ptrs[i] = is.readLong();

         // Read in the Values
         values = new Value[ph.getValueCount()];
         for ( int i = 0; i < values.length; i++ ) {
            short len = is.readShort();
            byte[] b = new byte[pfxLen+len];

            if ( pfxLen > 0 )
               System.arraycopy(pfxBytes, 0, b, 0, pfxLen);

            if ( len > 0 )
               is.read(b, pfxLen, len);
            values[i] = new Value(b);
         }

         cache.put(new Long(page.getPageNum()), this);
      }

      public synchronized void write(Transaction tx) throws DBException, IOException {
         ByteArrayOutputStream bos = new ByteArrayOutputStream(fileHeader.getWorkSize());
         DataOutputStream os = new DataOutputStream(bos);

         short pfxLen = ph.getPrefixLength();
         if ( pfxLen > 0 )
            prefix.streamTo(os);

         // Write out the pointers
         for ( int i = 0; i < ptrs.length; i++ )
            os.writeLong(ptrs[i]);

         // Write out the Values
         for ( int i = 0; i < values.length; i++ ) {
            Value v = values[i];
            byte[] b = v.getRawData();
            int pos = v.getOffset();
            int len = v.getLength();
            os.writeShort(len-pfxLen);
            if ( len-pfxLen > 0 )
               os.write(b, pos+pfxLen, len-pfxLen);
         }

         writeValue(tx, page, new Value(bos.toByteArray()));

         cache.put(new Long(page.getPageNum()), this);
      }

      private BTreeNode getChildNode(Transaction tx, int idx) throws IOException {
         boolean load;
         BTreeRootInfo loadNode;
         long loadPtr;
         synchronized ( this ) {
            if ( ph.getStatus() == BRANCH && idx >= 0 && idx < ptrs.length ) {
               load = true;
               loadNode = root;
               loadPtr = ptrs[idx];
            }
            else {
               load = false;
               loadNode = null;
               loadPtr = 0;
            }
         }
         if ( load )
            return getBTreeNode(tx, loadNode, loadPtr, this);
         else
            return null;
      }

      private synchronized long removeValue(Transaction tx, Value value) throws DBException, IOException {
         int idx = Arrays.binarySearch(values, value);
         switch ( ph.getStatus() ) {
            case BRANCH:
               idx = idx < 0 ? -(idx + 1) : idx + 1;
               return getChildNode(tx, idx).removeValue(tx, value);

            case LEAF:
               if ( idx < 0 )
                  throw new BTreeNotFoundException("Value '" + value.toString() + "' doesn't exist");
               else {
                  long oldPtr = ptrs[idx];

                  setValues(deleteArrayValue(values, idx));
                  setPointers(Array.deleteArrayLong(ptrs, idx));

                  write(tx);
                  return oldPtr;
               }

            default:
               throw new BTreeCorruptException("Invalid Page Type In removeValue");
         }
      }

      private boolean mustSplit() {
         // Only split if there is more than one value
         int valCount = values.length;
         if ( valCount > 1 ) {
            int sepLen = prefix.getLength();
            int ptrCount = ph.getStatus() == BRANCH ? valCount + 1 : valCount;
            int size = sepLen + (ptrCount * 8) + (valCount * 2) + 2;
            for ( int i = 0; i < valCount; i++ )
               size += values[i].getLength()-sepLen;

            return size >= fileHeader.getWorkSize();
         }
         else
            return false;
      }

      public synchronized long addValue(Transaction tx, Value value, long pointer) throws DBException, IOException {
         int idx = Arrays.binarySearch(values, value);

         switch ( ph.getStatus() ) {
            case BRANCH:
               idx = idx < 0 ? -(idx + 1) : idx + 1;
               return getChildNode(tx, idx).addValue(tx, value, pointer);

            case LEAF:
               if ( idx >= 0 ) {
                  // Value was found... Overwrite
                  long oldPtr = ptrs[idx];
                  ptrs[idx] = pointer;

                  setValues(values);
                  setPointers(ptrs);

                  write(tx);
                  return oldPtr;
               }
               else {
                  // Value was not found
                  idx = -(idx + 1);

                  setValues(insertArrayValue(values, value, idx));
                  setPointers(Array.insertArrayLong(ptrs, pointer, idx));

                  if ( mustSplit() )
                     split(tx);
                  else
                     write(tx);
               }
               return -1;

            default:
               throw new BTreeCorruptException("Invalid Page Type In addValue");
         }
      }

      private synchronized void promoteValue(Transaction tx, Value value, long rightPointer) throws DBException, IOException {
         int idx = Arrays.binarySearch(values, value);
         idx = idx < 0 ? -(idx + 1) : idx + 1;

         setValues(insertArrayValue(values, value, idx));
         setPointers(Array.insertArrayLong(ptrs, rightPointer, idx + 1));

         if ( mustSplit() )
            split(tx);
         else
            write(tx);
      }

      private Value getPrefix(Value value1, Value value2) {
         int idx = Math.abs(value1.compareTo(value2))-1;
         if ( idx > 0 ) {
            byte[] b = value2.getRawData();
            int pos = value2.getOffset();
            return new Value(b, pos, idx);
         }
         else
            return EmptyValue;
      }

      private Value getSeparator(Value value1, Value value2) {
         int idx = Math.abs(value1.compareTo(value2));
         byte[] b = value2.getRawData();
         int pos = value2.getOffset();
         return new Value(b, pos, idx);
      }

      private synchronized void split(Transaction tx) throws DBException, IOException {
         Value[] leftVals;
         Value[] rightVals;
         long[] leftPtrs;
         long[] rightPtrs;
         Value separator;

         short vc = ph.getValueCount();
         int pivot = vc / 2;

         // Split the node into two nodes
         switch ( ph.getStatus() ) {
            case BRANCH:
               leftVals = new Value[pivot];
               leftPtrs = new long[leftVals.length + 1];
               rightVals = new Value[vc - (pivot + 1)];
               rightPtrs = new long[rightVals.length + 1];

               System.arraycopy(values, 0, leftVals, 0, leftVals.length);
               System.arraycopy(ptrs, 0, leftPtrs, 0, leftPtrs.length);
               System.arraycopy(values, leftVals.length + 1, rightVals, 0, rightVals.length);
               System.arraycopy(ptrs, leftPtrs.length, rightPtrs, 0, rightPtrs.length);

               separator = values[leftVals.length];
               break;

            case LEAF:
               leftVals = new Value[pivot];
               leftPtrs = new long[leftVals.length];
               rightVals = new Value[vc - pivot];
               rightPtrs = new long[rightVals.length];

               System.arraycopy(values, 0, leftVals, 0, leftVals.length);
               System.arraycopy(ptrs, 0, leftPtrs, 0, leftPtrs.length);
               System.arraycopy(values, leftVals.length, rightVals, 0, rightVals.length);
               System.arraycopy(ptrs, leftPtrs.length, rightPtrs, 0, rightPtrs.length);

               separator = getSeparator(leftVals[leftVals.length - 1], rightVals[0]);
               break;

            default:
               throw new BTreeCorruptException("Invalid Page Type In split");
         }

         setValues(leftVals);
         setPointers(leftPtrs);

         // Promote the pivot to the parent branch
         if ( parent == null ) {
            // This can only happen if this is the root
            BTreeNode np = createBTreeNode(tx, root, BRANCH, null);

            BTreeNode rNode = createBTreeNode(tx, root, ph.getStatus(), np);
            rNode.setValues(rightVals);
            rNode.setPointers(rightPtrs);

            np.setValues(new Value[]{separator});
            np.setPointers(new long[]{page.getPageNum(), rNode.page.getPageNum()});

            parent = np;

            setRootNode(tx, np);

            write(tx);
            rNode.write(tx);
            np.write(tx);
         }
         else {
            BTreeNode rNode = createBTreeNode(tx, root, ph.getStatus(), parent);
            rNode.setValues(rightVals);
            rNode.setPointers(rightPtrs);

            write(tx);
            rNode.write(tx);
            parent.promoteValue(tx, separator, rNode.page.getPageNum());
         }
      }

      /////////////////////////////////////////////////////////////////

      public synchronized long findValue(Transaction tx, Value value) throws IOException, BTreeException {
         int idx = Arrays.binarySearch(values, value);

         switch ( ph.getStatus() ) {
            case BRANCH:
               idx = idx < 0 ? -(idx + 1) : idx + 1;
               return getChildNode(tx, idx).findValue(tx, value);

            case LEAF:
               if ( idx < 0 )
                  throw new BTreeNotFoundException("Value '" + value.toString() + "' doesn't exist");
               else
                  return ptrs[idx];

            default:
               throw new BTreeCorruptException("Invalid Page Type In findValue");
         }
      }

      private Value getIndexValue(Value value) {
         if ( compoundVals )
            return IndexValue.getIndexValue(value);
         else
            return value;
      }

      private Value getExtraValue(Value value) {
         if ( compoundVals )
            return IndexValue.getExtraValue(value);
         else
            return null;
      }

      // query is a BEAST of a method
      public synchronized void query(Transaction tx, IndexQuery query, BTreeCallback callback) throws IOException, BTreeException {
         if ( query != null && query.getOperator() != IndexQuery.ANY ) {
            Value[] qvals = query.getValues();
            int leftIdx = Arrays.binarySearch(values, qvals[0]);
            int rightIdx = qvals.length > 1 ? Arrays.binarySearch(values, qvals[qvals.length - 1]) : leftIdx;
            int op = query.getOperator();

            switch ( ph.getStatus() ) {
               case BRANCH:
                  leftIdx = leftIdx < 0 ? -(leftIdx + 1) : leftIdx + 1;
                  rightIdx = rightIdx < 0 ? -(rightIdx + 1) : rightIdx + 1;

                  switch ( op ) {
                     case IndexQuery.EQ:
                        if ( !compoundVals ) {
                           getChildNode(tx, leftIdx).query(tx, query, callback);
                           break;
                        }
                        // The fall through is intentional

                     case IndexQuery.BWX:
                     case IndexQuery.BW:
                     case IndexQuery.IN:
                     case IndexQuery.SW:
                        for ( int i = 0; i < ptrs.length; i++ )
                           if ( i >= leftIdx && i <= rightIdx )
                              getChildNode(tx, i).query(tx, query, callback);
                        break;

                     case IndexQuery.NEQ:
                        if ( !compoundVals ) {
                           for ( int i = 0; i < ptrs.length; i++ )
                              getChildNode(tx, i).query(tx, query, callback);
                           break;
                        }
                        // The fall through is intentional

                     case IndexQuery.NBWX:
                     case IndexQuery.NBW:
                     case IndexQuery.NIN:
                     case IndexQuery.NSW:
                        for ( int i = 0; i < ptrs.length; i++ )
                           if ( i <= leftIdx || i >= rightIdx )
                              getChildNode(tx, i).query(tx, query, callback);
                        break;

                     case IndexQuery.LT:
                     case IndexQuery.LTE:
                        for ( int i = 0; i < ptrs.length; i++ )
                           if ( i <= leftIdx )
                              getChildNode(tx, i).query(tx, query, callback);
                        break;

                     case IndexQuery.GT:
                     case IndexQuery.GTE:
                        for ( int i = 0; i < ptrs.length; i++ )
                           if ( i >= rightIdx )
                              getChildNode(tx, i).query(tx, query, callback);
                        break;

                     default:
                        for ( int i = 0; i < ptrs.length; i++ )
                           getChildNode(tx, i).query(tx, query, callback);
                        break;
                  }
                  break;

               case LEAF:
                  switch ( op ) {
                     case IndexQuery.EQ:
                        if ( !compoundVals ) {
                           if ( leftIdx >= 0 )
                              callback.indexInfo(values[leftIdx], null);
                           break;
                        }
                        // The fall through is intentional

                     case IndexQuery.BWX:
                     case IndexQuery.BW:
                     case IndexQuery.SW:
                     case IndexQuery.IN:
                        if ( leftIdx < 0 )
                           leftIdx = -(leftIdx + 1);
                        if ( rightIdx < 0 )
                           rightIdx = -(rightIdx + 1);
                        for ( int i = 0; i < ptrs.length; i++ ) {
                           Value testValue = getIndexValue(values[i]);
                           if ( i >= leftIdx && i <= rightIdx && query.testValue(testValue) ) {
                              Value extraValue = getExtraValue(values[i]);
                              callback.indexInfo(testValue, extraValue);
                           }
                        }
                        break;

                     case IndexQuery.NEQ:
                        if ( !compoundVals ) {
                           for ( int i = 0; i < ptrs.length; i++ )
                              if ( i != leftIdx )
                                 callback.indexInfo(values[i], null);
                           break;
                        }
                        // The fall through is intentional

                     case IndexQuery.NBWX:
                     case IndexQuery.NBW:
                     case IndexQuery.NSW:
                        if ( leftIdx < 0 )
                           leftIdx = -(leftIdx + 1);
                        if ( rightIdx < 0 )
                           rightIdx = -(rightIdx + 1);
                        for ( int i = 0; i < ptrs.length; i++ ) {
                           Value testValue = getIndexValue(values[i]);
                           if ( (i <= leftIdx || i >= rightIdx) && query.testValue(testValue) ) {
                              Value extraValue = getExtraValue(values[i]);
                              callback.indexInfo(testValue, extraValue);
                           }
                        }
                        break;

                     case IndexQuery.LT:
                     case IndexQuery.LTE:
                        if ( leftIdx < 0 )
                           leftIdx = -(leftIdx + 1);
                        for ( int i = 0; i < ptrs.length; i++ ) {
                           Value testValue = getIndexValue(values[i]);
                           if ( i <= leftIdx && query.testValue(testValue) ) {
                              Value extraValue = getExtraValue(values[i]);
                              callback.indexInfo(testValue, extraValue);
                           }
                        }
                        break;

                     case IndexQuery.GT:
                     case IndexQuery.GTE:
                        if ( rightIdx < 0 )
                           rightIdx = -(rightIdx + 1);
                        for ( int i = 0; i < ptrs.length; i++ ) {
                           Value testValue = getIndexValue(values[i]);
                           if ( i >= rightIdx && query.testValue(testValue) ) {
                              Value extraValue = getExtraValue(values[i]);
                              callback.indexInfo(testValue, extraValue);
                           }
                        }
                        break;

                     case IndexQuery.NIN:
                     default:
                        for ( int i = 0; i < ptrs.length; i++ ) {
                           Value testValue = getIndexValue(values[i]);
                           if ( query.testValue(testValue) ) {
                              Value extraValue = getExtraValue(values[i]);
                              callback.indexInfo(testValue, extraValue);
                           }
                        }
                        break;
                  }
                  break;

               default:
                  throw new BTreeCorruptException("Invalid Page Type In query");
            }

         }
         else {
            // No Query - Just Walk The Tree
            switch ( ph.getStatus() ) {
               case BRANCH:
                  for ( int i = 0; i < ptrs.length; i++ )
                     getChildNode(tx, i).query(tx, query, callback);
                  break;

               case LEAF:
                  for ( int i = 0; i < values.length; i++ ) {
                     Value testValue = getIndexValue(values[i]);
                     Value extraValue = getExtraValue(values[i]);
                     callback.indexInfo(testValue, extraValue);
                  }
                  break;

               default:
                  throw new BTreeCorruptException("Invalid Page Type In query");
            }
         }
      }
   }

   ////////////////////////////////////////////////////////////////////

   public FileHeader createFileHeader() {
      return new BTreeFileHeader();
   }

   public FileHeader createFileHeader(boolean read) throws IOException {
      return new BTreeFileHeader(read);
   }

   public FileHeader createFileHeader(long pageCount) {
      return new BTreeFileHeader(pageCount);
   }

   public FileHeader createFileHeader(long pageCount, int pageSize) {
      return new BTreeFileHeader(pageCount, pageSize);
   }

   public PageHeader createPageHeader() {
      return new BTreePageHeader();
   }


   /**
    * BTreeFileHeader
    */

   protected class BTreeFileHeader extends FileHeader {
      private long rootPage = 0;

      public BTreeFileHeader() {
      }

      public BTreeFileHeader(long pageCount) {
         super(pageCount);
      }

      public BTreeFileHeader(long pageCount, int pageSize) {
         super(pageCount, pageSize);
      }

      public BTreeFileHeader(boolean read) throws IOException {
         super(read);
      }

      public synchronized void read(RandomAccessFile raf) throws IOException {
         super.read(raf);
         rootPage = raf.readLong();
      }

      public synchronized void write(RandomAccessFile raf) throws IOException {
         super.write(raf);
         raf.writeLong(rootPage);
      }

      /** The root page of the storage tree */
      public synchronized final void setRootPage(long rootPage) {
         this.rootPage = rootPage;
         setDirty();
      }

      /** The root page of the storage tree */
      public synchronized final long getRootPage() {
         return rootPage;
      }
   }


   /**
    * BTreePageHeader
    */

   protected class BTreePageHeader extends PageHeader {
      private short valueCount = 0;
      private short prefixLength = 0;

      public BTreePageHeader() {
      }

      public BTreePageHeader(ByteBuffer buf) throws IOException {
         super(buf);
      }

      public synchronized void read(ByteBuffer buf) throws IOException {
         super.read(buf);

         if ( getStatus() == UNUSED )
            return;

         valueCount = buf.getShort();
         prefixLength = buf.getShort();
      }

      public synchronized void write(ByteBuffer buf) throws IOException {
         super.write(buf);
         buf.putShort(valueCount);
         buf.putShort(prefixLength);
      }

      /** The number of values stored by this page */
      public synchronized final void setValueCount(short valueCount) {
         this.valueCount = valueCount;
         setDirty();
      }

      /** The number of values stored by this page */
      public synchronized final short getValueCount() {
         return valueCount;
      }

      /** The number of pointers stored by this page */
      public synchronized final short getPointerCount() {
         if ( getStatus() == BRANCH )
            return (short)(valueCount + 1);
         else
            return valueCount;
      }

      public synchronized final short getPrefixLength() {
         return prefixLength;
      }

      public synchronized final void setPrefixLength(short prefixLength) {
         this.prefixLength = prefixLength;
         setDirty();
      }
   }
}

