/*
   Copyright (c) 2008 Health Market Science, Inc.
   
   This library is free software; you can redistribute it and/or
   modify it under the terms of the GNU Lesser General Public
   License as published by the Free Software Foundation; either
   version 2.1 of the License, or (at your option) any later version.
   
   This library 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
  Lesser General Public License for more details.
  
  You should have received a copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
  USA
  
  You can contact Health Market Science at info@healthmarketscience.com
  or at the following address:
  
  Health Market Science
  2700 Horizon Drive
  Suite 200
  King of Prussia, PA 19406
  */
  
  package com.healthmarketscience.jackcess;
  
  import java.io.IOException;
  import java.lang.ref.Reference;
  import java.lang.ref.SoftReference;
  import java.util.AbstractList;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.RandomAccess;
  
  
  import static com.healthmarketscience.jackcess.Index.*;
  
  /**
   * Manager of the index pages for a BigIndex.
   * @author James Ahlborn
   */
  public class IndexPageCache
  {
    private enum UpdateType {
      ADD, REMOVE, REPLACE;
    }
  
    /** the index whose pages this cache is managing */
    private final BigIndex _index;
    /** the root page for the index */
    private DataPageMain _rootPage;
    /** the currently loaded pages for this index, pageNumber -> page */
    private final Map<Integer, DataPageMain> _dataPages =
      new HashMap<Integer, DataPageMain>();
    /** the currently modified index pages */
    private final List<CacheDataPage> _modifiedPages =
      new ArrayList<CacheDataPage>();
    
    public IndexPageCache(BigIndex index) {
      _index = index;
    }
  
    public BigIndex getIndex() {
      return _index;
    }
    
    public PageChannel getPageChannel() {
      return getIndex().getPageChannel();
    }
  
    /**
     * Sets the root page for this index, must be called before normal usage.
     *
     * @param pageNumber the root page number
     */
    public void setRootPageNumber(int pageNumber) throws IOException {
      _rootPage = getDataPage(pageNumber);
      // root page has no parent
      _rootPage.initParentPage(INVALID_INDEX_PAGE_NUMBER, false);
    }
    
    /**
     * Writes any outstanding changes for this index to the file.
     */
    public void write()
      throws IOException
    {
      // first discard any empty pages
      handleEmptyPages();
      // next, handle any necessary page splitting
      preparePagesForWriting();
      // finally, write all the modified pages (which are not being deleted)
     writeDataPages();
   }
 
   /**
    * Handles any modified pages which are empty as the first pass during a
    * {@link #write} call.  All empty pages are removed from the _modifiedPages
    * collection by this method.
    */
   private void handleEmptyPages() throws IOException
   {
     for(Iterator<CacheDataPage> iter = _modifiedPages.iterator();
         iter.hasNext(); ) {
       CacheDataPage cacheDataPage = iter.next();
       if(cacheDataPage._extra._entryView.isEmpty()) {
         if(!cacheDataPage._main.isRoot()) {
           deleteDataPage(cacheDataPage);
         } else {
           writeDataPage(cacheDataPage);
         }
         iter.remove();
       }
     }
   }
   
   /**
    * Prepares any non-empty modified pages for writing as the second pass
    * during a {@link #write} call.  Updates entry prefixes, promotes/demotes
    * tail pages, and splits pages as needed.
    */
   private void preparePagesForWriting() throws IOException
   {
     boolean splitPages = false;
     int maxPageEntrySize = getIndex().getMaxPageEntrySize();
 
     // we need to continue looping through all the pages until we do not split
     // any pages (because a split may cascade up the tree)
     do {
       splitPages = false;
 
       // we might be adding to this list while iterating, so we can't use an
       // iterator
       for(int i = 0; i < _modifiedPages.size(); ++i) {
 
         CacheDataPage cacheDataPage = _modifiedPages.get(i);
 
         if(!cacheDataPage.isLeaf()) {
           // see if we need to update any child tail status
           DataPageMain dpMain = cacheDataPage._main;
           int size = cacheDataPage._extra._entryView.size();
           if(dpMain.hasChildTail()) {
             if(size == 1) {
               demoteTail(cacheDataPage);
             } 
           } else {
             if(size > 1) {
               promoteTail(cacheDataPage);
             }
           }
         }
         
         // look for pages with more entries than can fit on a page
         if(cacheDataPage.getTotalEntrySize() > maxPageEntrySize) {
 
           // make sure the prefix is up-to-date (this may have gotten
           // discarded by one of the update entry methods)
           cacheDataPage._extra.updateEntryPrefix();
 
           // now, see if the page will fit when compressed
           if(cacheDataPage.getCompressedEntrySize() > maxPageEntrySize) {
             // need to split this page
             splitPages = true;
             splitDataPage(cacheDataPage);
           }
         }
       }
       
     } while(splitPages);
   }
 
   /**
    * Writes any non-empty modified pages as the last pass during a
    * {@link #write} call.  Clears the _modifiedPages collection when finised.
    */
   private void writeDataPages() throws IOException
   {
     for(CacheDataPage cacheDataPage : _modifiedPages) {
       if(cacheDataPage._extra._entryView.isEmpty()) {
         throw new IllegalStateException("Unexpected empty page " +
                                         cacheDataPage);
       }
       writeDataPage(cacheDataPage);
     }
     _modifiedPages.clear();
   }
 
   /**
    * Returns a CacheDataPage for the given page number, may be {@code null} if
    * the given page number is invalid.  Loads the given page if necessary.
    */
   public CacheDataPage getCacheDataPage(Integer pageNumber)
     throws IOException
   {
     DataPageMain main = getDataPage(pageNumber);
     return((main != null) ? new CacheDataPage(main) : null);
   }
   
   /**
    * Returns a DataPageMain for the given page number, may be {@code null} if
    * the given page number is invalid.  Loads the given page if necessary.
    */
   private DataPageMain getDataPage(Integer pageNumber)
     throws IOException
   {
     DataPageMain dataPage = _dataPages.get(pageNumber);
     if((dataPage == null) && (pageNumber > INVALID_INDEX_PAGE_NUMBER)) {
       dataPage = readDataPage(pageNumber)._main;
       _dataPages.put(pageNumber, dataPage);
     }
     return dataPage;
   }
 
   /**
    * Writes the given index page to the file.
    */
   private void writeDataPage(CacheDataPage cacheDataPage)
     throws IOException
   {
     getIndex().writeDataPage(cacheDataPage);
 
     // lastly, mark the page as no longer modified
     cacheDataPage._extra._modified = false;    
   }
   
   /**
    * Deletes the given index page from the file (clears the page).
    */
   private void deleteDataPage(CacheDataPage cacheDataPage)
     throws IOException
   {
     // free this database page
     getPageChannel().deallocatePage(cacheDataPage._main._pageNumber);
 
     // discard from our cache
     _dataPages.remove(cacheDataPage._main._pageNumber);
     
     // lastly, mark the page as no longer modified
     cacheDataPage._extra._modified = false;    
   }
   
   /**
    * Reads the given index page from the file.
    */
   private CacheDataPage readDataPage(Integer pageNumber)
     throws IOException
   {
     DataPageMain dataPage = new DataPageMain(pageNumber);
     DataPageExtra extra = new DataPageExtra();
     CacheDataPage cacheDataPage = new CacheDataPage(dataPage, extra);
     getIndex().readDataPage(cacheDataPage);
 
     // associate the extra info with the main data page
     dataPage.setExtra(extra);
     
     return cacheDataPage;
   }  
 
   /**
    * Removes the entry with the given index from the given page.
    *
    * @param cacheDataPage the page from which to remove the entry
    * @param entryIdx the index of the entry to remove
    */
   private void removeEntry(CacheDataPage cacheDataPage, int entryIdx)
     throws IOException
   {
     updateEntry(cacheDataPage, entryIdx, null, UpdateType.REMOVE);
   }
 
   /**
    * Adds the entry to the given page at the given index.
    *
    * @param cacheDataPage the page to which to add the entry
    * @param entryIdx the index at which to add the entry
    * @param newEntry the entry to add
    */
   private void addEntry(CacheDataPage cacheDataPage,
                         int entryIdx,
                         Entry newEntry)
     throws IOException
   {
     updateEntry(cacheDataPage, entryIdx, newEntry, UpdateType.ADD);
   }
   
   /**
    * Updates the entries on the given page according to the given updateType.
    *
    * @param cacheDataPage the page to update
    * @param entryIdx the index at which to add/remove/replace the entry
    * @param newEntry the entry to add/replace
    * @param upType the type of update to make
    */
   private void updateEntry(CacheDataPage cacheDataPage,
                            int entryIdx,
                            Entry newEntry,
                            UpdateType upType)
     throws IOException
   {
     DataPageMain dpMain = cacheDataPage._main;
     DataPageExtra dpExtra = cacheDataPage._extra;
 
     if(newEntry != null) {
       validateEntryForPage(dpMain, newEntry);
     }
 
     // note, it's slightly ucky, but we need to load the parent page before we
     // start mucking with our entries because our parent may use our entries.
     CacheDataPage parentDataPage = (!dpMain.isRoot() ?
                                     new CacheDataPage(dpMain.getParentPage()) :
                                     null);
     
     Entry oldLastEntry = dpExtra._entryView.getLast();
     Entry oldEntry = null;
     int entrySizeDiff = 0;
 
     switch(upType) {
     case ADD:
       dpExtra._entryView.add(entryIdx, newEntry);
       entrySizeDiff += newEntry.size();
       break;
 
     case REPLACE:
       oldEntry = dpExtra._entryView.set(entryIdx, newEntry);
       entrySizeDiff += newEntry.size() - oldEntry.size();
       break;
 
     case REMOVE: {
       oldEntry = dpExtra._entryView.remove(entryIdx);
       entrySizeDiff -= oldEntry.size();
       break;
     }
     default:
       throw new RuntimeException("unknown update type " + upType);
     }
 
     boolean updateLast = (oldLastEntry != dpExtra._entryView.getLast());
     
     // child tail entry updates do not modify the page
     if(!updateLast || !dpMain.hasChildTail()) {
       dpExtra._totalEntrySize += entrySizeDiff;
       setModified(cacheDataPage);
 
       // for now, just clear the prefix, we'll fix it later
       dpExtra._entryPrefix = EMPTY_PREFIX;
     }
 
     if(dpExtra._entryView.isEmpty()) {
       // this page is dead
       removeDataPage(parentDataPage, cacheDataPage, oldLastEntry);
       return;
     }
 
     // determine if we need to update our parent page 
     if(!updateLast || dpMain.isRoot()) {
       // no parent
       return;
     }
 
     // the update to the last entry needs to be propagated to our parent
     replaceParentEntry(parentDataPage, cacheDataPage, oldLastEntry);
   }
 
   /**
    * Removes an index page which has become empty.  If this page is the root
    * page, just clears it.
    *
    * @param parentDataPage the parent of the removed page
    * @param cacheDataPage the page to remove
    * @param oldLastEntry the last entry for this page (before it was removed)
    */
   private void removeDataPage(CacheDataPage parentDataPage,
                               CacheDataPage cacheDataPage,
                               Entry oldLastEntry)
     throws IOException
   {
     DataPageMain dpMain = cacheDataPage._main;
     DataPageExtra dpExtra = cacheDataPage._extra;
 
     if(dpMain.hasChildTail()) {
       throw new IllegalStateException("Still has child tail?");
     }
 
     if(dpExtra._totalEntrySize != 0) {
       throw new IllegalStateException("Empty page but size is not 0? " +
                                       dpExtra._totalEntrySize + ", " +
                                       cacheDataPage);
     }
     
     if(dpMain.isRoot()) {
       // clear out this page (we don't actually remove it)
       dpExtra._entryPrefix = EMPTY_PREFIX;
       // when the root page becomes empty, it becomes a leaf page again
       dpMain._leaf = true;
       return;
     }
 
     // remove this page from its parent page
     updateParentEntry(parentDataPage, cacheDataPage, oldLastEntry, null,
                       UpdateType.REMOVE);
 
     // remove this page from any next/prev pages
     removeFromPeers(cacheDataPage);
   }
 
   /**
    * Removes a now empty index page from its next and previous peers.
    *
    * @param cacheDataPage the page to remove
    */
   private void removeFromPeers(CacheDataPage cacheDataPage)
     throws IOException
   {
     DataPageMain dpMain = cacheDataPage._main;
 
     Integer prevPageNumber = dpMain._prevPageNumber;
     Integer nextPageNumber = dpMain._nextPageNumber;
     
     DataPageMain prevMain = dpMain.getPrevPage();
     if(prevMain != null) {
       setModified(new CacheDataPage(prevMain));
       prevMain._nextPageNumber = nextPageNumber;
     }
 
     DataPageMain nextMain = dpMain.getNextPage();
     if(nextMain != null) {
       setModified(new CacheDataPage(nextMain));
       nextMain._prevPageNumber = prevPageNumber;
     }
   }
 
   /**
    * Adds an entry for the given child page to the given parent page.
    *
    * @param parentDataPage the parent page to which to add the entry
    * @param childDataPage the child from which to get the entry to add
    */
   private void addParentEntry(CacheDataPage parentDataPage,
                               CacheDataPage childDataPage)
     throws IOException
   {
     DataPageExtra childExtra = childDataPage._extra;
     updateParentEntry(parentDataPage, childDataPage, null,
                       childExtra._entryView.getLast(), UpdateType.ADD);
   }
   
   /**
    * Replaces the entry for the given child page in the given parent page.
    *
    * @param parentDataPage the parent page in which to replace the entry
    * @param childDataPage the child for which the entry is being replaced
    * @param oldEntry the old child entry for the child page
    */
   private void replaceParentEntry(CacheDataPage parentDataPage,
                                   CacheDataPage childDataPage,
                                   Entry oldEntry)
     throws IOException
   {
     DataPageExtra childExtra = childDataPage._extra;
     updateParentEntry(parentDataPage, childDataPage, oldEntry,
                       childExtra._entryView.getLast(), UpdateType.REPLACE);
   }
   
   /**
    * Updates the entry for the given child page in the given parent page
    * according to the given updateType.
    *
    * @param parentDataPage the parent page in which to update the entry
    * @param childDataPage the child for which the entry is being updated
    * @param oldEntry the old child entry to remove/replace
    * @param newEntry the new child entry to replace/add
    * @param upType the type of update to make
    */
   private void updateParentEntry(CacheDataPage parentDataPage,
                                  CacheDataPage childDataPage,
                                  Entry oldEntry, Entry newEntry,
                                  UpdateType upType)
     throws IOException
   {
     DataPageMain childMain = childDataPage._main;
     DataPageExtra parentExtra = parentDataPage._extra;
 
     if(childMain.isTail() && (upType != UpdateType.REMOVE)) {
       // for add or replace, update the child tail info before updating the
       // parent entries
       updateParentTail(parentDataPage, childDataPage, upType);
     }
 
     if(oldEntry != null) {
       oldEntry = oldEntry.asNodeEntry(childMain._pageNumber);
     }
     if(newEntry != null) {
       newEntry = newEntry.asNodeEntry(childMain._pageNumber);
     }
 
     boolean expectFound = true;
     int idx = 0;
     
     switch(upType) {
     case ADD:
       expectFound = false;
       idx = parentExtra._entryView.find(newEntry);
       break;
 
     case REPLACE:
     case REMOVE:
       idx = parentExtra._entryView.find(oldEntry);
       break;
     
     default:
       throw new RuntimeException("unknown update type " + upType);
     }
         
     if(idx < 0) {
       if(expectFound) {
         throw new IllegalStateException(
             "Could not find child entry in parent; childEntry " + oldEntry +
             "; parent " + parentDataPage);
       }
       idx = missingIndexToInsertionPoint(idx);
     } else {
       if(!expectFound) {
         throw new IllegalStateException(
             "Unexpectedly found child entry in parent; childEntry " +
             newEntry + "; parent " + parentDataPage);
       }
     }
     updateEntry(parentDataPage, idx, newEntry, upType);
 
     if(childMain.isTail() && (upType == UpdateType.REMOVE)) {
       // for remove, update the child tail info after updating the parent
       // entries
       updateParentTail(parentDataPage, childDataPage, upType);
     }
   }
 
   /**
    * Updates the child tail info in the given parent page according to the
    * given updateType.
    *
    * @param parentDataPage the parent page in which to update the child tail
    * @param childDataPage the child to add/replace
    * @param upType the type of update to make
    */
   private void updateParentTail(CacheDataPage parentDataPage,
                                 CacheDataPage childDataPage,
                                 UpdateType upType)
     throws IOException
   {
     DataPageMain parentMain = parentDataPage._main;
 
     int newChildTailPageNumber =
       ((upType == UpdateType.REMOVE) ?
        INVALID_INDEX_PAGE_NUMBER :
        childDataPage._main._pageNumber);
     if(!parentMain.isChildTailPageNumber(newChildTailPageNumber)) {
       setModified(parentDataPage);
       parentMain._childTailPageNumber = newChildTailPageNumber;
     }
   }
   
   /**
    * Verifies that the given entry type (node/leaf) is valid for the given
    * page (node/leaf).
    *
    * @param dpMain the page to which the entry will be added
    * @param entry the entry being added
    * @throws IllegalStateException if the entry type does not match the page
    *         type
    */
   private void validateEntryForPage(DataPageMain dpMain, Entry entry) {
     if(dpMain._leaf != entry.isLeafEntry()) {
       throw new IllegalStateException(
           "Trying to update page with wrong entry type; pageLeaf " +
           dpMain._leaf + ", entryLeaf " + entry.isLeafEntry());
     }
   }
 
   /**
    * Splits an index page which has too many entries on it.
    *
    * @param origDataPage the page to split
    */
   private void splitDataPage(CacheDataPage origDataPage)
     throws IOException
   {
     DataPageMain origMain = origDataPage._main;
     DataPageExtra origExtra = origDataPage._extra;
 
     setModified(origDataPage);
     
     int numEntries = origExtra._entries.size();
     if(numEntries < 2) {
       throw new IllegalStateException(
           "Cannot split page with less than 2 entries " + origDataPage);
     }
     
     if(origMain.isRoot()) {
       // we can't split the root page directly, so we need to put another page
       // between the root page and its sub-pages, and then split that page.
       CacheDataPage newDataPage = nestRootDataPage(origDataPage);
 
       // now, split this new page instead
       origDataPage = newDataPage;
       origMain = newDataPage._main;
       origExtra = newDataPage._extra;
     }
 
     // note, it's slightly ucky, but we need to load the parent page before we
     // start mucking with our entries because our parent may use our entries.
     DataPageMain parentMain = origMain.getParentPage();
     CacheDataPage parentDataPage = new CacheDataPage(parentMain);
     
     // note, there are many, many ways this could be improved/tweaked.  for
     // now, we just want it to be functional...
     // so, we will naively move half the entries from one page to a new page.
 
     CacheDataPage newDataPage = allocateNewCacheDataPage(
         parentMain._pageNumber, origMain._leaf);
     DataPageMain newMain = newDataPage._main;
     DataPageExtra newExtra = newDataPage._extra;
     
     List<Entry> headEntries =
       origExtra._entries.subList(0, ((numEntries + 1) / 2));
 
     // move first half of the entries from old page to new page (so we do not
     // need to muck with any tail entries)
     for(Entry headEntry : headEntries) {
       newExtra._totalEntrySize += headEntry.size();
       newExtra._entries.add(headEntry);
     }
     newExtra.setEntryView(newMain);
 
     // remove the moved entries from the old page
     headEntries.clear();
     origExtra._entryPrefix = EMPTY_PREFIX;
     origExtra._totalEntrySize -= newExtra._totalEntrySize;
 
     // insert this new page between the old page and any previous page
     addToPeersBefore(newDataPage, origDataPage);
 
     if(!newMain._leaf) {
       // reparent the children pages of the new page
       reparentChildren(newDataPage);
 
       // if the children of this page are also node pages, then the next/prev
       // links should not cross parent boundaries (the leaf pages are linked
       // from beginning to end, but child node pages are only linked within
       // the same parent)
       DataPageMain childMain = newMain.getChildPage(
           newExtra._entryView.getLast());
       if(!childMain._leaf) {
         separateFromNextPeer(new CacheDataPage(childMain));
       }
     }
 
     // lastly, we need to add the new page to the parent page's entries
     addParentEntry(parentDataPage, newDataPage);
   }
 
   /**
    * Copies the current root page info into a new page and nests this page
    * under the root page.  This must be done when the root page needs to be
    * split.
    *
    * @param rootDataPage the root data page
    * 
    * @return the newly created page nested under the root page
    */
   private CacheDataPage nestRootDataPage(CacheDataPage rootDataPage)
     throws IOException
   {
     DataPageMain rootMain = rootDataPage._main;
     DataPageExtra rootExtra = rootDataPage._extra;
 
     if(!rootMain.isRoot()) {
       throw new IllegalArgumentException("should be called with root, duh");
     }
     
     CacheDataPage newDataPage =
       allocateNewCacheDataPage(rootMain._pageNumber, rootMain._leaf);
     DataPageMain newMain = newDataPage._main;
     DataPageExtra newExtra = newDataPage._extra;
 
     // move entries to new page
     newMain._childTailPageNumber = rootMain._childTailPageNumber;
     newExtra._entries = rootExtra._entries;
     newExtra._entryPrefix = rootExtra._entryPrefix;
     newExtra._totalEntrySize = rootExtra._totalEntrySize;
     newExtra.setEntryView(newMain);
 
     if(!newMain._leaf) {
       // we need to re-parent all the child pages
       reparentChildren(newDataPage);
     }
       
     // clear the root page
     rootMain._leaf = false;
     rootMain._childTailPageNumber = INVALID_INDEX_PAGE_NUMBER;
     rootExtra._entries = new ArrayList<Entry>();
     rootExtra._entryPrefix = EMPTY_PREFIX;
     rootExtra._totalEntrySize = 0;
     rootExtra.setEntryView(rootMain);
 
     // add the new page as the first child of the root page
     addParentEntry(rootDataPage, newDataPage);
 
     return newDataPage;
   }
   
   /**
    * Allocates a new index page with the given parent page and type.
    *
    * @param parentPageNumber the parent page for the new page
    * @param isLeaf whether or not the new page is a leaf page
    *
    * @return the newly created page
    */
   private CacheDataPage allocateNewCacheDataPage(Integer parentPageNumber,
                                                  boolean isLeaf)
     throws IOException
   {
     DataPageMain dpMain = new DataPageMain(getPageChannel().allocateNewPage());
     DataPageExtra dpExtra = new DataPageExtra();
     dpMain.initParentPage(parentPageNumber, false);
     dpMain._leaf = isLeaf;
     dpMain._prevPageNumber = INVALID_INDEX_PAGE_NUMBER;
     dpMain._nextPageNumber = INVALID_INDEX_PAGE_NUMBER;
     dpMain._childTailPageNumber = INVALID_INDEX_PAGE_NUMBER;
     dpExtra._entries = new ArrayList<Entry>();
     dpExtra._entryPrefix = EMPTY_PREFIX;
     dpMain.setExtra(dpExtra);
 
     // add to our page cache
     _dataPages.put(dpMain._pageNumber, dpMain);
 
     // needs to be written out
     CacheDataPage cacheDataPage = new CacheDataPage(dpMain, dpExtra);
     setModified(cacheDataPage);
 
     return cacheDataPage;
   }
 
   /**
    * Inserts the new page as a peer between the given original page and any
    * previous peer page.
    *
    * @param newDataPage the new index page
    * @param origDataPage the current index page
    */
   private void addToPeersBefore(CacheDataPage newDataPage,
                                 CacheDataPage origDataPage)
     throws IOException
   {
     DataPageMain origMain = origDataPage._main;
     DataPageMain newMain = newDataPage._main;
 
     DataPageMain prevMain = origMain.getPrevPage();
 
     newMain._nextPageNumber = origMain._pageNumber;
     newMain._prevPageNumber = origMain._prevPageNumber;
     origMain._prevPageNumber = newMain._pageNumber;
     
     if(prevMain != null) {
       setModified(new CacheDataPage(prevMain));
       prevMain._nextPageNumber = newMain._pageNumber;
     }
   }
 
   /**
    * Separates the given index page from any next peer page.
    *
    * @param cacheDataPage the index page to be separated
    */
   private void separateFromNextPeer(CacheDataPage cacheDataPage)
     throws IOException
   {
     DataPageMain dpMain = cacheDataPage._main;
 
     setModified(cacheDataPage);
 
     DataPageMain nextMain = dpMain.getNextPage();
     setModified(new CacheDataPage(nextMain));
 
     nextMain._prevPageNumber = INVALID_INDEX_PAGE_NUMBER;
     dpMain._nextPageNumber = INVALID_INDEX_PAGE_NUMBER;
   }
   
   /**
    * Sets the parent info for the children of the given page to the given
    * page.
    *
    * @param cacheDataPage the page whose children need to be updated
    */
   private void reparentChildren(CacheDataPage cacheDataPage)
     throws IOException
   {
     DataPageMain dpMain = cacheDataPage._main;
     DataPageExtra dpExtra = cacheDataPage._extra;
 
     // note, the "parent" page number is not actually persisted, so we do not
     // need to mark any updated pages as modified.  for the same reason, we
     // don't need to load the pages if not already loaded
     for(Entry entry : dpExtra._entryView) {
       Integer childPageNumber = entry.getSubPageNumber();
       DataPageMain childMain = _dataPages.get(childPageNumber);
       if(childMain != null) {
         childMain.setParentPage(dpMain._pageNumber,
                                 dpMain.isChildTailPageNumber(childPageNumber));
       }
     }
   }
 
   /**
    * Makes the tail entry of the given page a normal entry on that page, done
    * when there is only one entry left on a page, and it is the tail.
    *
    * @param cacheDataPage the page whose tail must be updated
    */
   private void demoteTail(CacheDataPage cacheDataPage)
     throws IOException
   {
     // there's only one entry on the page, and it's the tail.  make it a
     // normal entry
     DataPageMain dpMain = cacheDataPage._main;
     DataPageExtra dpExtra = cacheDataPage._extra;
 
     setModified(cacheDataPage);
     
     DataPageMain tailMain = dpMain.getChildTailPage();
     CacheDataPage tailDataPage = new CacheDataPage(tailMain);
 
     // move the tail entry to the last normal entry
     updateParentTail(cacheDataPage, tailDataPage, UpdateType.REMOVE);
     Entry tailEntry = dpExtra._entryView.demoteTail();
     dpExtra._totalEntrySize += tailEntry.size();
     dpExtra._entryPrefix = EMPTY_PREFIX;
     
     tailMain.setParentPage(dpMain._pageNumber, false);
   }
   
   /**
    * Makes the last normal entry of the given page the tail entry on that
    * page, done when there are multiple entries on a page and no tail entry.
    *
    * @param cacheDataPage the page whose tail must be updated
    */
   private void promoteTail(CacheDataPage cacheDataPage)
     throws IOException
   {
     // there's not tail currently on this page, make last entry a tail
     DataPageMain dpMain = cacheDataPage._main;
     DataPageExtra dpExtra = cacheDataPage._extra;
 
     setModified(cacheDataPage);
     
     DataPageMain lastMain = dpMain.getChildPage(dpExtra._entryView.getLast());
     CacheDataPage lastDataPage = new CacheDataPage(lastMain);
 
     // move the "last" normal entry to the tail entry
     updateParentTail(cacheDataPage, lastDataPage, UpdateType.ADD);
     Entry lastEntry = dpExtra._entryView.promoteTail();
     dpExtra._totalEntrySize -= lastEntry.size();
     dpExtra._entryPrefix = EMPTY_PREFIX;
 
     lastMain.setParentPage(dpMain._pageNumber, true);
   }
   
   /**
    * Finds the index page on which the given entry does or should reside.
    *
    * @param e the entry to find
    */
   public CacheDataPage findCacheDataPage(Entry e)
     throws IOException
   {
     DataPageMain curPage = _rootPage;
     while(true) {
 
       if(curPage._leaf) {
         // nowhere to go from here
         return new CacheDataPage(curPage);
       }
       
       DataPageExtra extra = curPage.getExtra();
 
       // need to descend
       int idx = extra._entryView.find(e);
       if(idx < 0) {
         idx = missingIndexToInsertionPoint(idx);
         if(idx == extra._entryView.size()) {
           // just move to last child page
           --idx;
         }
       }
 
       Entry nodeEntry = extra._entryView.get(idx);
       curPage = curPage.getChildPage(nodeEntry);
     }
   }
 
   /**
    * Marks the given index page as modified and saves it for writing, if
    * necessary (if the page is already marked, does nothing).
    *
    * @param cacheDataPage the modified index page
    */
   private void setModified(CacheDataPage cacheDataPage)
   {
     if(!cacheDataPage._extra._modified) {
       _modifiedPages.add(cacheDataPage);
       cacheDataPage._extra._modified = true;
     }
   }
 
   /**
    * Finds the valid entry prefix given the first/last entries on an index
    * page.
    *
    * @param e1 the first entry on the page
    * @param e2 the last entry on the page
    *
    * @return a valid entry prefix for the page
    */
   private static byte[] findCommonPrefix(Entry e1, Entry e2)
   {
     byte[] b1 = e1.getEntryBytes();
     byte[] b2 = e2.getEntryBytes();
     
     int maxLen = b1.length;
     byte[] prefix = b1;
     if(b1.length > b2.length) {
       maxLen = b2.length;
       prefix = b2;
     }
     
     int len = 0;
     while((len < maxLen) && (b1[len] == b2[len])) {
       ++len;
     }
     
     if(len < prefix.length) {
       if(len == 0) {
         return EMPTY_PREFIX;
       }
       
       // need new prefix
       byte[] tmpPrefix = new byte[len];
       System.arraycopy(prefix, 0, tmpPrefix, 0, len);
       prefix = tmpPrefix;
     }
 
     return prefix;
   }
 
   /**
    * Used by unit tests to validate the internal status of the index.
    */
   void validate() throws IOException {
     for(DataPageMain dpMain : _dataPages.values()) {
       DataPageExtra dpExtra = dpMain.getExtra();
       validateEntries(dpExtra);
       validateChildren(dpMain, dpExtra);
       validatePeers(dpMain);
     }
   }
 
   /**
    * Validates the entries for an index page
    *
    * @param dpExtra the entries to validate
    */
   private void validateEntries(DataPageExtra dpExtra) throws IOException {
     int entrySize = 0;
     Entry prevEntry = Index.FIRST_ENTRY;
     for(Entry e : dpExtra._entries) {
       entrySize += e.size();
       if(prevEntry.compareTo(e) >= 0) {
         throw new IOException("Unexpected order in index entries, " +
                               prevEntry + " >= " + e);
       }
       prevEntry = e;
     }
     if(entrySize != dpExtra._totalEntrySize) {
       throw new IllegalStateException("Expected size " + entrySize +
                                       " but was " + dpExtra._totalEntrySize);
     }
   }
 
   /**
    * Validates the children for an index page
    *
    * @param dpMain the index page
    * @param dpExtra the child entries to validate
    */
   private void validateChildren(DataPageMain dpMain,
                                 DataPageExtra dpExtra) throws IOException {
     int childTailPageNumber = dpMain._childTailPageNumber;
     if(dpMain._leaf) {
       if(childTailPageNumber != INVALID_INDEX_PAGE_NUMBER) {
         throw new IllegalStateException("Leaf page has tail " + dpMain);
       }
       return;
     }
     if((dpExtra._entryView.size() == 1) && dpMain.hasChildTail()) {
       throw new IllegalStateException("Single child is tail " + dpMain);
     }
     for(Entry e : dpExtra._entryView) {
       validateEntryForPage(dpMain, e);
       Integer subPageNumber = e.getSubPageNumber();
       DataPageMain childMain = _dataPages.get(subPageNumber);
       if(childMain != null) {
         if(childMain._parentPageNumber != null) {
           if((int)childMain._parentPageNumber != dpMain._pageNumber) {
             throw new IllegalStateException("Child's parent is incorrect " +
                                             childMain);
           }
           boolean expectTail = ((int)subPageNumber == childTailPageNumber);
           if(expectTail != childMain._tail) {
             throw new IllegalStateException("Child tail status incorrect " +
                                             childMain);
           }
         }
         Entry lastEntry = childMain.getExtra()._entryView.getLast();
         if(e.compareTo(lastEntry) != 0) {
           throw new IllegalStateException("Invalid entry " + e +
                                           " but child is " + lastEntry);
         }
       }
     }
   }
 
   /**
    * Validates the peer pages for an index page.
    *
    * @param dpMain the index page
    */
   private void validatePeers(DataPageMain dpMain) throws IOException {
     DataPageMain prevMain = _dataPages.get(dpMain._prevPageNumber);
     if(prevMain != null) {
       if((int)prevMain._nextPageNumber != dpMain._pageNumber) {
         throw new IllegalStateException("Prev page " + prevMain +
                                         " does not ref " + dpMain);
       }
       validatePeerStatus(dpMain, prevMain);
     }
     
     DataPageMain nextMain = _dataPages.get(dpMain._nextPageNumber);
     if(nextMain != null) {
       if((int)nextMain._prevPageNumber != dpMain._pageNumber) {
         throw new IllegalStateException("Next page " + nextMain +
                                         " does not ref " + dpMain);
       }
       validatePeerStatus(dpMain, nextMain);
     }
   }
 
   /**
    * Validates the given peer page against the given index page
    *
    * @param dpMain the index page
    * @param peerMain the peer index page
    */
   private void validatePeerStatus(DataPageMain dpMain, DataPageMain peerMain)
     throws IOException
   {
     if(dpMain._leaf != peerMain._leaf) {
       throw new IllegalStateException("Mismatched peer status " +
                                       dpMain._leaf + " " + peerMain._leaf);
     }
     if(!dpMain._leaf) {
       if((dpMain._parentPageNumber != null) &&
          (peerMain._parentPageNumber != null) &&
          ((int)dpMain._parentPageNumber != (int)peerMain._parentPageNumber)) {
         throw new IllegalStateException("Mismatched node parents " +
                                         dpMain._parentPageNumber + " " +
                                         peerMain._parentPageNumber);
       }
     }
   }
   
   /**
    * Dumps the given index page to a StringBuilder
    *
    * @param rtn the StringBuilder to update
    * @param dpMain the index page to dump
    */
   private void dumpPage(StringBuilder rtn, DataPageMain dpMain) {
     try {
       CacheDataPage cacheDataPage = new CacheDataPage(dpMain);
       rtn.append(cacheDataPage).append("\n");
       if(!dpMain._leaf) {
         for(Entry e : cacheDataPage._extra._entryView) {
           DataPageMain childMain = dpMain.getChildPage(e);
           dumpPage(rtn, childMain);
         }
       }
     } catch(IOException e) {
       rtn.append("Page[" + dpMain._pageNumber + "]: " + e);
     }
   }
   
   @Override
   public String toString() {
     if(_rootPage == null) {
       return "Cache: (uninitialized)";
     }
     
     StringBuilder rtn = new StringBuilder("Cache: \n");
     dumpPage(rtn, _rootPage);
     return rtn.toString();
   }
 
   /**
    * Keeps track of the main info for an index page.
    */
   private class DataPageMain
   {
     public final int _pageNumber;
     public Integer _prevPageNumber;
     public Integer _nextPageNumber;
     public Integer _childTailPageNumber;
     public Integer _parentPageNumber;
     public boolean _leaf;
     public boolean _tail;
     private Reference<DataPageExtra> _extra;
 
     private DataPageMain(int pageNumber) {
       _pageNumber = pageNumber;
     }
 
     public IndexPageCache getCache() {
       return IndexPageCache.this;
     }
     
     public boolean isRoot() {
       return(this == _rootPage);
     }
     
     public boolean isTail() throws IOException
     {
       resolveParent();
       return _tail;
     }
 
     public boolean hasChildTail() {
       return((int)_childTailPageNumber != INVALID_INDEX_PAGE_NUMBER);
     }
 
     public boolean isChildTailPageNumber(int pageNumber) {
       return((int)_childTailPageNumber == pageNumber);
     }
     
     public DataPageMain getParentPage() throws IOException
     {
       resolveParent();
       return IndexPageCache.this.getDataPage(_parentPageNumber);
     }
 
     public void initParentPage(Integer parentPageNumber, boolean isTail) {
       // only set if not already set
       if(_parentPageNumber == null) {
         setParentPage(parentPageNumber, isTail);
       }
     }
     
     public void setParentPage(Integer parentPageNumber, boolean isTail) {
       _parentPageNumber = parentPageNumber;
       _tail = isTail;
     }
 
     public DataPageMain getPrevPage() throws IOException
     {
       return IndexPageCache.this.getDataPage(_prevPageNumber);
     }
     
     public DataPageMain getNextPage() throws IOException
     {
       return IndexPageCache.this.getDataPage(_nextPageNumber);
     }
     
     public DataPageMain getChildPage(Entry e) throws IOException
     {
       Integer childPageNumber = e.getSubPageNumber();
       return getChildPage(childPageNumber,
                           isChildTailPageNumber(childPageNumber));
     }
     
     public DataPageMain getChildTailPage() throws IOException
     {
       return getChildPage(_childTailPageNumber, true);
     }
 
     /**
      * Returns a child page for the given page number, updating its parent
      * info if necessary.
      */
     private DataPageMain getChildPage(Integer childPageNumber, boolean isTail)
       throws IOException
     {
       DataPageMain child = getDataPage(childPageNumber);
       if(child != null) {
         // set the parent info for this child (if necessary)
         child.initParentPage(_pageNumber, isTail);
       }
       return child;
     }
     
     public DataPageExtra getExtra() throws IOException
     {
       DataPageExtra extra = _extra.get();
       if(extra == null) {
         extra = readDataPage(_pageNumber)._extra;
         setExtra(extra);
       }
       
       return extra;
     }
 
     public void setExtra(DataPageExtra extra) throws IOException
     {
       extra.setEntryView(this);
       _extra = new SoftReference<DataPageExtra>(extra);
     }
 
     private void resolveParent() throws IOException {
       if(_parentPageNumber == null) {
         // the act of searching for the last entry should resolve any parent
         // pages along the path
         findCacheDataPage(getExtra()._entryView.getLast());
         if(_parentPageNumber == null) {
           throw new IllegalStateException("Parent was not resolved");
         }
       }
     }
 
     @Override
     public String toString() {
       return (_leaf ? "Leaf" : "Node") + "DPMain[" + _pageNumber +
         "] " + _prevPageNumber + ", " + _nextPageNumber + ", (" +
         _childTailPageNumber + ")";
     }
   }
 
   /**
    * Keeps track of the extra info for an index page.  This info (if
    * unmodified) may be re-read from disk as necessary.
    */
   private static class DataPageExtra
   {
     /** sorted collection of index entries.  this is kept in a list instead of
         a SortedSet because the SortedSet has lame traversal utilities */
     public List<Entry> _entries;
     public EntryListView _entryView;
     public byte[] _entryPrefix;
     public int _totalEntrySize;
     public boolean _modified;
 
     private DataPageExtra()
     {
     }
 
     public void setEntryView(DataPageMain main) throws IOException {
       _entryView = new EntryListView(main, this);
     }
     
     public void updateEntryPrefix() {
       if(_entryPrefix.length == 0) {
         // prefix is only related to *real* entries, tail not included
         _entryPrefix = findCommonPrefix(_entries.get(0),
                                         _entries.get(_entries.size() - 1));
       }
     }
     
     @Override
     public String toString() {
       return "DPExtra: " + _entryView;
     }    
   }
 
   /**
    * IndexPageCache implementation of an Index {@link DataPage}.
    */
   public static final class CacheDataPage
     extends Index.DataPage
   {
     public final DataPageMain _main;
     public final DataPageExtra _extra;
 
     private CacheDataPage(DataPageMain dataPage) throws IOExc