/*
   Copyright (c) 2005 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.ByteArrayOutputStream;
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.nio.ByteOrder;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import static com.healthmarketscience.jackcess.IndexCodes.*;
  
  
  /**
   * Access table index
   * @author Tim McCune
   */
  public abstract class Index implements Comparable<Index> {
    
    protected static final Log LOG = LogFactory.getLog(Index.class);
  
    /** special entry which is less than any other entry */
    public static final Entry FIRST_ENTRY =
      createSpecialEntry(RowId.FIRST_ROW_ID);
    
    /** special entry which is greater than any other entry */
    public static final Entry LAST_ENTRY =
      createSpecialEntry(RowId.LAST_ROW_ID);
    
    protected static final int INVALID_INDEX_PAGE_NUMBER = 0;          
    
    /** Max number of columns in an index */
    private static final int MAX_COLUMNS = 10;
    
    protected static final byte[] EMPTY_PREFIX = new byte[0];
  
    private static final short COLUMN_UNUSED = -1;
  
    private static final byte ASCENDING_COLUMN_FLAG = (byte)0x01;
  
    private static final byte UNIQUE_INDEX_FLAG = (byte)0x01;
    private static final byte IGNORE_NULLS_INDEX_FLAG = (byte)0x02;
  
    /** index type for primary key indexes */
    private static final byte PRIMARY_KEY_INDEX_TYPE = (byte)1;
    
    /** index type for foreign key indexes */
    private static final byte FOREIGN_KEY_INDEX_TYPE = (byte)2;
  
    private static final int MAX_TEXT_INDEX_CHAR_LENGTH =
      (JetFormat.TEXT_FIELD_MAX_LENGTH / JetFormat.TEXT_FIELD_UNIT_SIZE);
    
    /** type attributes for Entries which simplify comparisons */
    public enum EntryType {
      /** comparable type indicating this Entry should always compare less than
          valid RowIds */
      ALWAYS_FIRST,
      /** comparable type indicating this Entry should always compare less than
          other valid entries with equal entryBytes */
      FIRST_VALID,
      /** comparable type indicating this RowId should always compare
          normally */
      NORMAL,
      /** comparable type indicating this Entry should always compare greater
         than other valid entries with equal entryBytes */
     LAST_VALID,
     /** comparable type indicating this Entry should always compare greater
         than valid RowIds */
     ALWAYS_LAST;
   }
   
   static final Comparator<byte[]> BYTE_CODE_COMPARATOR =
     new Comparator<byte[]>() {
       public int compare(byte[] left, byte[] right) {
         if(left == right) {
           return 0;
         }
         if(left == null) {
           return -1;
         }
         if(right == null) {
           return 1;
         }
 
         int len = Math.min(left.length, right.length);
         int pos = 0;
         while((pos < len) && (left[pos] == right[pos])) {
           ++pos;
         }
         if(pos < len) {
           return ((ByteUtil.asUnsignedByte(left[pos]) <
                    ByteUtil.asUnsignedByte(right[pos])) ? -1 : 1);
         }
         return ((left.length < right.length) ? -1 :
                 ((left.length > right.length) ? 1 : 0));
       }
     };
         
   
   /** owning table */
   private final Table _table;
   /** Page number of the root index data */
   private int _rootPageNumber;
   /** offset within the tableDefinition buffer of the uniqueEntryCount for
       this index */
   private final int _uniqueEntryCountOffset;
   /** The number of unique entries which have been added to this index.  note,
       however, that it is never decremented, only incremented (as observed in
       Access). */
   private int _uniqueEntryCount;
   /** List of columns and flags */
   private final List<ColumnDescriptor> _columns =
     new ArrayList<ColumnDescriptor>();
   /** 0-based index number */
   private int _indexNumber;
   /** flags for this index */
   private byte _indexFlags;
   /** the type of the index */
   private byte _indexType;
   /** Index name */
   private String _name;
   /** <code>true</code> if the index entries have been initialized,
       <code>false</code> otherwise */
   private boolean _initialized;
   /** modification count for the table, keeps cursors up-to-date */
   private int _modCount;
   /** temp buffer used to read/write the index pages */
   private final TempBufferHolder _indexBufferH =
     TempBufferHolder.newHolder(TempBufferHolder.Type.SOFT, true);
   /** max size for all the entries written to a given index data page */
   private final int _maxPageEntrySize;
   /** FIXME, for now, we can't write multi-page indexes or indexes using the funky primary key compression scheme */
   boolean _readOnly;
   
   protected Index(Table table, int uniqueEntryCount,
                   int uniqueEntryCountOffset)
   {
     _table  = table;
     _uniqueEntryCount = uniqueEntryCount;
     _uniqueEntryCountOffset = uniqueEntryCountOffset;
     _maxPageEntrySize = calcMaxPageEntrySize(_table.getFormat());
   }
 
   public Table getTable() {
     return _table;
   }
   
   public JetFormat getFormat() {
     return getTable().getFormat();
   }
 
   public PageChannel getPageChannel() {
     return getTable().getPageChannel();
   }
 
   public void setIndexNumber(int indexNumber) {
     _indexNumber = indexNumber;
   }
 
   public int getIndexNumber() {
     return _indexNumber;
   }
 
   public void setIndexType(byte indexType) {
     _indexType = indexType;
   }
 
   public byte getIndexFlags() {
     return _indexFlags;
   }
   
   public int getUniqueEntryCount() {
     return _uniqueEntryCount;
   }
 
   public int getUniqueEntryCountOffset() {
     return _uniqueEntryCountOffset;
   }
 
   public String getName() {
     return _name;
   }
   
   public void setName(String name) {
     _name = name;
   }
 
   public boolean isPrimaryKey() {
     return _indexType == PRIMARY_KEY_INDEX_TYPE;
   }
 
   public boolean isForeignKey() {
     return _indexType == FOREIGN_KEY_INDEX_TYPE;
   }
 
   /**
    * Whether or not {@code null} values are actually recorded in the index.
    */
   public boolean shouldIgnoreNulls() {
     return((_indexFlags & IGNORE_NULLS_INDEX_FLAG) != 0);
   }
 
   /**
    * Whether or not index entries must be unique.
    * <p>
    * Some notes about uniqueness:
    * <ul>
    * <li>Access does not seem to consider multiple {@code null} entries
    *     invalid for a unique index</li>
    * <li>text indexes collapse case, and Access seems to compare <b>only</b>
    *     the index entry bytes, therefore two strings which differ only in
    *     case <i>will violate</i> the unique constraint</li>
    * </ul>
    */
   public boolean isUnique() {
     return(isPrimaryKey() || ((_indexFlags & UNIQUE_INDEX_FLAG) != 0));
   }
   
   /**
    * Returns the Columns for this index (unmodifiable)
    */
   public List<ColumnDescriptor> getColumns() {
     return Collections.unmodifiableList(_columns);
   }
 
   /**
    * Whether or not the complete index state has been read.
    */
   public boolean isInitialized() {
     return _initialized;
   }
 
   protected int getRootPageNumber() {
     return _rootPageNumber;
   }
 
   protected void setReadOnly() {
     _readOnly = true;
   }
 
   protected int getMaxPageEntrySize() {
     return _maxPageEntrySize;
   }
 
   /**
    * Returns the number of index entries in the index.  Only called by unit
    * tests.
    * <p>
    * Forces index initialization.
    */
   protected int getEntryCount()
     throws IOException
   {
     initialize();
     EntryCursor cursor = cursor();
     Entry endEntry = cursor.getLastEntry();
     int count = 0;
     while(!endEntry.equals(cursor.getNextEntry())) {
       ++count;
     }
     return count;
   }
   
   /**
    * Forces initialization of this index (actual parsing of index pages).
    * normally, the index will not be initialized until the entries are
    * actually needed.
    */
   public void initialize() throws IOException {
     if(!_initialized) {
       readIndexEntries();
       _initialized = true;
     }
   }
 
   /**
    * Writes the current index state to the database.
    * <p>
    * Forces index initialization.
    */
   public void update() throws IOException
   {
     // make sure we've parsed the entries
     initialize();
     
     if(_readOnly) {
       throw new UnsupportedOperationException(
           "FIXME cannot write indexes of this type yet, see Database javadoc for info on enabling large index support");
     }
     updateImpl();
   }
 
   /**
    * Read the index info from a tableBuffer
    * @param tableBuffer table definition buffer to read from initial info
    * @param availableColumns Columns that this index may use
    */
   public void read(ByteBuffer tableBuffer, List<Column> availableColumns)
     throws IOException
   {
     for (int i = 0; i < MAX_COLUMNS; i++) {
       short columnNumber = tableBuffer.getShort();
       byte colFlags = tableBuffer.get();
       if (columnNumber != COLUMN_UNUSED) {
         // find the desired column by column number (which is not necessarily
         // the same as the column index)
         Column idxCol = null;
         for(Column col : availableColumns) {
           if(col.getColumnNumber() == columnNumber) {
             idxCol = col;
             break;
           }
         }
         if(idxCol == null) {
           throw new IOException("Could not find column with number "
                                 + columnNumber + " for index " + getName());
         }
         _columns.add(newColumnDescriptor(idxCol, colFlags));
       }
     }
     tableBuffer.getInt(); //Forward past Unknown
     _rootPageNumber = tableBuffer.getInt();
     tableBuffer.getInt(); //Forward past Unknown
     _indexFlags = tableBuffer.get();
     tableBuffer.position(tableBuffer.position() + 5);  //Forward past other stuff
   }
 
   /**
    * Adds a row to this index
    * <p>
    * Forces index initialization.
    * 
    * @param row Row to add
    * @param rowId rowId of the row to be added
    */
   public void addRow(Object[] row, RowId rowId)
     throws IOException
   {
     int nullCount = countNullValues(row);
     boolean isNullEntry = (nullCount == _columns.size());
     if(shouldIgnoreNulls() && isNullEntry) {
       // nothing to do
       return;
     }
     if(isPrimaryKey() && (nullCount > 0)) {
       throw new IOException("Null value found in row " + Arrays.asList(row)
                             + " for primary key index " + this);
     }
     
     // make sure we've parsed the entries
     initialize();
 
     Entry newEntry = new Entry(createEntryBytes(row), rowId);
     if(addEntry(newEntry, isNullEntry, row)) {
       ++_modCount;
     } else {
       LOG.warn("Added duplicate index entry " + newEntry + " for row: " +
                Arrays.asList(row));
     }
   }
 
   /**
    * Adds an entry to the correct index dataPage, maintaining the order.
    */
   private boolean addEntry(Entry newEntry, boolean isNullEntry, Object[] row)
     throws IOException
   {
     DataPage dataPage = findDataPage(newEntry);
     int idx = dataPage.findEntry(newEntry);
     if(idx < 0) {
       // this is a new entry
       idx = missingIndexToInsertionPoint(idx);
 
       Position newPos = new Position(dataPage, idx, newEntry, true);
       Position nextPos = getNextPosition(newPos);
       Position prevPos = getPreviousPosition(newPos);
       
       // determine if the addition of this entry would break the uniqueness
       // constraint.  See isUnique() for some notes about uniqueness as
       // defined by Access.
       boolean isDupeEntry =
         (((nextPos != null) &&
           newEntry.equalsEntryBytes(nextPos.getEntry())) ||
           ((prevPos != null) &&
            newEntry.equalsEntryBytes(prevPos.getEntry())));
       if(isUnique() && !isNullEntry && isDupeEntry)
       {
         throw new IOException(
             "New row " + Arrays.asList(row) +
             " violates uniqueness constraint for index " + this);
       }
 
       if(!isDupeEntry) {
         ++_uniqueEntryCount;
       }
 
       dataPage.addEntry(idx, newEntry);
       return true;
     }
     return false;
   }
   
   /**
    * Removes a row from this index
    * <p>
    * Forces index initialization.
    * 
    * @param row Row to remove
    * @param rowId rowId of the row to be removed
    */
   public void deleteRow(Object[] row, RowId rowId)
     throws IOException
   {
     int nullCount = countNullValues(row);
     if(shouldIgnoreNulls() && (nullCount == _columns.size())) {
       // nothing to do
       return;
     }
     
     // make sure we've parsed the entries
     initialize();
 
     Entry oldEntry = new Entry(createEntryBytes(row), rowId);
     if(removeEntry(oldEntry)) {
       ++_modCount;
     } else {
       LOG.warn("Failed removing index entry " + oldEntry + " for row: " +
                Arrays.asList(row));
     }
   }
 
   /**
    * Removes an entry from the relevant index dataPage, maintaining the order.
    * Will search by RowId if entry is not found (in case a partial entry was
    * provided).
    */
   private boolean removeEntry(Entry oldEntry)
     throws IOException
   {
     DataPage dataPage = findDataPage(oldEntry);
     int idx = dataPage.findEntry(oldEntry);
     boolean doRemove = false;
     if(idx < 0) {
       // the caller may have only read some of the row data, if this is the
       // case, just search for the page/row numbers
       // FIXME, we could force caller to get relevant values?
       EntryCursor cursor = cursor();
       Position tmpPos = null;
       Position endPos = cursor._lastPos;
       while(!endPos.equals(
                 tmpPos = cursor.getAnotherPosition(Cursor.MOVE_FORWARD))) {
         if(tmpPos.getEntry().getRowId().equals(oldEntry.getRowId())) {
           dataPage = tmpPos.getDataPage();
           idx = tmpPos.getIndex();
           doRemove = true;
           break;
         }
       }
     } else {
       doRemove = true;
     }
 
     if(doRemove) {
       // found it!
       dataPage.removeEntry(idx);
     }
     
     return doRemove;
   }
       
   /**
    * Gets a new cursor for this index.
    * <p>
    * Forces index initialization.
    */
   public EntryCursor cursor()
     throws IOException
   {
     return cursor(null, true, null, true);
   }
   
   /**
    * Gets a new cursor for this index, narrowed to the range defined by the
    * given startRow and endRow.
    * <p>
    * Forces index initialization.
    * 
    * @param startRow the first row of data for the cursor, or {@code null} for
    *                 the first entry
    * @param startInclusive whether or not startRow is inclusive or exclusive
    * @param endRow the last row of data for the cursor, or {@code null} for
    *               the last entry
    * @param endInclusive whether or not endRow is inclusive or exclusive
    */
   public EntryCursor cursor(Object[] startRow,
                             boolean startInclusive,
                             Object[] endRow,
                             boolean endInclusive)
     throws IOException
   {
     initialize();
     Entry startEntry = FIRST_ENTRY;
     byte[] startEntryBytes = null;
     if(startRow != null) {
       startEntryBytes = createEntryBytes(startRow);
       startEntry = new Entry(startEntryBytes,
                              (startInclusive ?
                               RowId.FIRST_ROW_ID : RowId.LAST_ROW_ID));
     }
     Entry endEntry = LAST_ENTRY;
     if(endRow != null) {
       // reuse startEntryBytes if startRow and endRow are same array.  this is
       // common for "lookup" code
       byte[] endEntryBytes = ((startRow == endRow) ?
                               startEntryBytes :
                               createEntryBytes(endRow));
       endEntry = new Entry(endEntryBytes,
                            (endInclusive ?
                             RowId.LAST_ROW_ID : RowId.FIRST_ROW_ID));
     }
     return new EntryCursor(findEntryPosition(startEntry),
                            findEntryPosition(endEntry));
   }
 
   private Position findEntryPosition(Entry entry)
     throws IOException
   {
     DataPage dataPage = findDataPage(entry);
     int idx = dataPage.findEntry(entry);
     boolean between = false;
     if(idx < 0) {
       // given entry was not found exactly.  our current position is now
       // really between two indexes, but we cannot support that as an integer
       // value, so we set a flag instead
       idx = missingIndexToInsertionPoint(idx);
       between = true;
     }
     return new Position(dataPage, idx, entry, between);
   }
 
   private Position getNextPosition(Position curPos)
     throws IOException
   {
     // get the next index (between-ness is handled internally)
     int nextIdx = curPos.getNextIndex();
     Position nextPos = null;
     if(nextIdx < curPos.getDataPage().getEntries().size()) {
       nextPos = new Position(curPos.getDataPage(), nextIdx);
     } else {
       int nextPageNumber = curPos.getDataPage().getNextPageNumber();
       DataPage nextDataPage = null;
       while(nextPageNumber != INVALID_INDEX_PAGE_NUMBER) {
         DataPage dp = getDataPage(nextPageNumber);
         if(!dp.isEmpty()) {
           nextDataPage = dp;
           break;
         }
         nextPageNumber = dp.getNextPageNumber();
       }
       if(nextDataPage != null) {
         nextPos = new Position(nextDataPage, 0);
       }
     }
     return nextPos;
   }
 
   /**
    * Returns the Position before the given one, or {@code null} if none.
    */
   private Position getPreviousPosition(Position curPos)
     throws IOException
   {
     // get the previous index (between-ness is handled internally)
     int prevIdx = curPos.getPrevIndex();
     Position prevPos = null;
     if(prevIdx >= 0) {
       prevPos = new Position(curPos.getDataPage(), prevIdx);
     } else {
       int prevPageNumber = curPos.getDataPage().getPrevPageNumber();
       DataPage prevDataPage = null;
       while(prevPageNumber != INVALID_INDEX_PAGE_NUMBER) {
         DataPage dp = getDataPage(prevPageNumber);
         if(!dp.isEmpty()) {
           prevDataPage = dp;
           break;
         }
         prevPageNumber = dp.getPrevPageNumber();
       }
       if(prevDataPage != null) {
         prevPos = new Position(prevDataPage,
                                (prevDataPage.getEntries().size() - 1));
       }
     }
     return prevPos;
   }
 
   /**
    * Returns the valid insertion point for an index indicating a missing
    * entry.
    */
   protected static int missingIndexToInsertionPoint(int idx) {
     return -(idx + 1);
   }
 
   /**
    * Constructs an array of values appropriate for this index from the given
    * column values, expected to match the columns for this index.
    * @return the appropriate sparse array of data
    * @throws IllegalArgumentException if the wrong number of values are
    *         provided
    */
   public Object[] constructIndexRowFromEntry(Object... values)
   {
     if(values.length != _columns.size()) {
       throw new IllegalArgumentException(
           "Wrong number of column values given " + values.length +
           ", expected " + _columns.size());
     }
     int valIdx = 0;
     Object[] idxRow = new Object[getTable().getColumnCount()];
     for(ColumnDescriptor col : _columns) {
       idxRow[col.getColumnIndex()] = values[valIdx++];
     }
     return idxRow;
   }
     
   /**
    * Constructs an array of values appropriate for this index from the given
    * column value.
    * @return the appropriate sparse array of data or {@code null} if not all
    *         columns for this index were provided
    */
   public Object[] constructIndexRow(String colName, Object value)
   {
     return constructIndexRow(Collections.singletonMap(colName, value));
   }
   
   /**
    * Constructs an array of values appropriate for this index from the given
    * column values.
    * @return the appropriate sparse array of data or {@code null} if not all
    *         columns for this index were provided
    */
   public Object[] constructIndexRow(Map<String,Object> row)
   {
     for(ColumnDescriptor col : _columns) {
       if(!row.containsKey(col.getName())) {
         return null;
       }
     }
 
     Object[] idxRow = new Object[getTable().getColumnCount()];
     for(ColumnDescriptor col : _columns) {
       idxRow[col.getColumnIndex()] = row.get(col.getName());
     }
     return idxRow;
   }  
 
   @Override
   public String toString() {
     StringBuilder rtn = new StringBuilder();
     rtn.append("\tName: (" + _table.getName() + ") " + _name);
     rtn.append("\n\tNumber: " + _indexNumber);
     rtn.append("\n\tPage number: " + _rootPageNumber);
     rtn.append("\n\tIs Primary Key: " + isPrimaryKey());
     rtn.append("\n\tIs Foreign Key: " + isForeignKey());
     rtn.append("\n\tIs Unique: " + isUnique());
     rtn.append("\n\tIgnore Nulls: " + shouldIgnoreNulls());
     rtn.append("\n\tColumns: " + _columns);
     rtn.append("\n\tInitialized: " + _initialized);
     if(_initialized) {
       try {
         rtn.append("\n\tEntryCount: " + getEntryCount());
       } catch(IOException e) {
         throw new RuntimeException(e);
       }
     }
     rtn.append("\n\n");
     return rtn.toString();
   }
   
   public int compareTo(Index other) {
     if (_indexNumber > other.getIndexNumber()) {
       return 1;
     } else if (_indexNumber < other.getIndexNumber()) {
       return -1;
     } else {
       return 0;
     }
   }
 
   /**
    * Write the given index page out to a buffer
    */
   protected void writeDataPage(DataPage dataPage)
     throws IOException
   {
     if(dataPage.getCompressedEntrySize() > _maxPageEntrySize) {
       if(this instanceof SimpleIndex) {
         throw new UnsupportedOperationException(
             "FIXME cannot write large index yet, see Database javadoc for info on enabling large index support");
       }
       throw new IllegalStateException("data page is too large");
     }
     
     ByteBuffer buffer = _indexBufferH.getPageBuffer(getPageChannel());
     buffer.put(dataPage.isLeaf() ?
                PageTypes.INDEX_LEAF :
                PageTypes.INDEX_NODE );  //Page type
     buffer.put((byte) 0x01);  //Unknown
     buffer.putShort((short) 0); //Free space
     buffer.putInt(getTable().getTableDefPageNumber());
 
     buffer.putInt(0); //Unknown
     buffer.putInt(dataPage.getPrevPageNumber()); //Prev page
     buffer.putInt(dataPage.getNextPageNumber()); //Next page
     buffer.putInt(dataPage.getChildTailPageNumber()); //ChildTail page
 
     byte[] entryPrefix = dataPage.getEntryPrefix();
     buffer.putShort((short) entryPrefix.length); // entry prefix byte count
     buffer.put((byte) 0); //Unknown
 
     byte[] entryMask = new byte[getFormat().SIZE_INDEX_ENTRY_MASK];
     // first entry includes the prefix
     int totalSize = entryPrefix.length;
     for(Entry entry : dataPage.getEntries()) {
       totalSize += (entry.size() - entryPrefix.length);
       int idx = totalSize / 8;
       entryMask[idx] |= (1 << (totalSize % 8));
     }
     buffer.put(entryMask);
 
     // first entry includes the prefix
     buffer.put(entryPrefix);
     
     for(Entry entry : dataPage.getEntries()) {
       entry.write(buffer, entryPrefix);
     }
 
     // update free space
     buffer.putShort(2, (short) (getFormat().PAGE_SIZE - buffer.position()));
 
     getPageChannel().writePage(buffer, dataPage.getPageNumber());
   }
 
   /**
    * Reads an index page, populating the correct collection based on the page
    * type (node or leaf).
    */
   protected void readDataPage(DataPage dataPage)
     throws IOException
   {
     ByteBuffer buffer = _indexBufferH.getPageBuffer(getPageChannel());
     getPageChannel().readPage(buffer, dataPage.getPageNumber());
 
     boolean isLeaf = isLeafPage(buffer);
     dataPage.setLeaf(isLeaf);
 
     // note, "header" data is in LITTLE_ENDIAN format, entry data is in
     // BIG_ENDIAN format
     int entryPrefixLength = ByteUtil.getUnsignedShort(
         buffer, getFormat().OFFSET_INDEX_COMPRESSED_BYTE_COUNT);
     int entryMaskLength = getFormat().SIZE_INDEX_ENTRY_MASK;
     int entryMaskPos = getFormat().OFFSET_INDEX_ENTRY_MASK;
     int entryPos = entryMaskPos + entryMaskLength;
     int lastStart = 0;
     int totalEntrySize = 0;
     byte[] entryPrefix = null;
     List<Entry> entries = new ArrayList<Entry>();
     TempBufferHolder tmpEntryBufferH =
       TempBufferHolder.newHolder(TempBufferHolder.Type.HARD, true,
                                  ByteOrder.BIG_ENDIAN);
 
     Entry prevEntry = FIRST_ENTRY;
     for (int i = 0; i < entryMaskLength; i++) {
       byte entryMask = buffer.get(entryMaskPos + i);
       for (int j = 0; j < 8; j++) {
         if ((entryMask & (1 << j)) != 0) {
           int length = (i * 8) + j - lastStart;
           buffer.position(entryPos + lastStart);
 
           // determine if we can read straight from the index page (if no
           // entryPrefix).  otherwise, create temp buf with complete entry.
           ByteBuffer curEntryBuffer = buffer;
           int curEntryLen = length;
           if(entryPrefix != null) {
             curEntryBuffer = getTempEntryBuffer(
                 buffer, length, entryPrefix, tmpEntryBufferH);
             curEntryLen += entryPrefix.length;
           }
           totalEntrySize += curEntryLen;
 
           Entry entry = newEntry(curEntryBuffer, curEntryLen, isLeaf);
           if(prevEntry.compareTo(entry) >= 0) {
             throw new IOException("Unexpected order in index entries, " +
                                   prevEntry + " >= " + entry);
           }
           
           entries.add(entry);
 
           if((entries.size() == 1) && (entryPrefixLength > 0)) {
             // read any shared entry prefix
             entryPrefix = new byte[entryPrefixLength];
             buffer.position(entryPos + lastStart);
             buffer.get(entryPrefix);
           }
 
           lastStart += length;
           prevEntry = entry;
         }
       }
     }
 
     dataPage.setEntryPrefix(entryPrefix != null ? entryPrefix : EMPTY_PREFIX);
     dataPage.setEntries(entries);
     dataPage.setTotalEntrySize(totalEntrySize);
     
     int prevPageNumber = buffer.getInt(getFormat().OFFSET_PREV_INDEX_PAGE);
     int nextPageNumber = buffer.getInt(getFormat().OFFSET_NEXT_INDEX_PAGE);
     int childTailPageNumber =
       buffer.getInt(getFormat().OFFSET_CHILD_TAIL_INDEX_PAGE);
 
     dataPage.setPrevPageNumber(prevPageNumber);
     dataPage.setNextPageNumber(nextPageNumber);
     dataPage.setChildTailPageNumber(childTailPageNumber);
   }
 
   /**
    * Returns a new Entry of the correct type for the given data and page type.
    */
   private Entry newEntry(ByteBuffer buffer, int entryLength, boolean isLeaf)
     throws IOException
   {
     if(isLeaf) {
       return new Entry(buffer, entryLength);
     }
     return new NodeEntry(buffer, entryLength);
   }
 
   /**
    * Returns an entry buffer containing the relevant data for an entry given
    * the valuePrefix.
    */
   private ByteBuffer getTempEntryBuffer(
       ByteBuffer indexPage, int entryLen, byte[] valuePrefix,
       TempBufferHolder tmpEntryBufferH)
   {
     ByteBuffer tmpEntryBuffer = tmpEntryBufferH.getBuffer(
         getPageChannel(), valuePrefix.length + entryLen);
 
     // combine valuePrefix and rest of entry from indexPage, then prep for
     // reading
     tmpEntryBuffer.put(valuePrefix);
     tmpEntryBuffer.put(indexPage.array(), indexPage.position(), entryLen);
     tmpEntryBuffer.flip();
     
     return tmpEntryBuffer;
   }
     
   /**
    * Determines if the given index page is a leaf or node page.
    */
   private static boolean isLeafPage(ByteBuffer buffer)
     throws IOException
   {
     byte pageType = buffer.get(0);
     if(pageType == PageTypes.INDEX_LEAF) {
       return true;
     } else if(pageType == PageTypes.INDEX_NODE) {
       return false;
     }
     throw new IOException("Unexpected page type " + pageType);
   }
   
   /**
    * Determines the number of {@code null} values for this index from the
    * given row.
    */
   private int countNullValues(Object[] values)
   {
     if(values == null) {
       return _columns.size();
     }
     
     // annoyingly, the values array could come from different sources, one
     // of which will make it a different size than the other.  we need to
     // handle both situations.
     int nullCount = 0;
     for(ColumnDescriptor col : _columns) {
       Object value = values[col.getColumnIndex()];
       if(col.isNullValue(value)) {
         ++nullCount;
       }
     }
     
     return nullCount;
   }
 
   /**
    * Creates the entry bytes for a row of values.
    */
   private byte[] createEntryBytes(Object[] values) throws IOException
   {
     if(values == null) {
       return null;
     }
     
     ByteArrayOutputStream bout = new ByteArrayOutputStream();
     
     // annoyingly, the values array could come from different sources, one
     // of which will make it a different size than the other.  we need to
     // handle both situations.
     for(ColumnDescriptor col : _columns) {
       Object value = values[col.getColumnIndex()];
       col.writeValue(value, bout);
     }
     
     return bout.toByteArray();
   }  
   
   /**
    * Writes the current index state to the database.  Index has already been
    * initialized.
    */
   protected abstract void updateImpl() throws IOException;
   
   /**
    * Reads the actual index entries.
    */
   protected abstract void readIndexEntries()
     throws IOException;
 
   /**
    * Finds the data page for the given entry.
    */
   protected abstract DataPage findDataPage(Entry entry)
     throws IOException;
   
   /**
    * Gets the data page for the pageNumber.
    */
   protected abstract DataPage getDataPage(int pageNumber)
     throws IOException;
   
   /**
    * Flips the first bit in the byte at the given index.
    */
   private static byte[] flipFirstBitInByte(byte[] value, int index)
   {
     value[index] = (byte)(value[index] ^ 0x80);
 
     return value;
   }
 
   /**
    * Flips all the bits in the byte array.
    */
   private static byte[] flipBytes(byte[] value) {
     for(int i = 0; i < value.length; ++i) {
       value[i] = (byte)(~value[i]);
     } 
     return value;
   }
 
   /**
    * Writes the value of the given column type to a byte array and returns it.
    */
   private static byte[] encodeNumberColumnValue(Object value, Column column)
     throws IOException
   {
     // always write in big endian order
     return column.write(value, 0, ByteOrder.BIG_ENDIAN).array();
   }    
 
   /**
    * Converts an index value for a text column into the entry value (which
    * is based on a variety of nifty codes).
    */
   private static void writeNonNullIndexTextValue(
       Object value, ByteArrayOutputStream bout, boolean isAscending)
     throws IOException
   {
     // first, convert to string
     String str = Column.toCharSequence(value).toString();
 
     // all text columns (including memos) are only indexed up to the max
     // number of chars in a VARCHAR column
     if(str.length() > MAX_TEXT_INDEX_CHAR_LENGTH) {
       str = str.substring(0, MAX_TEXT_INDEX_CHAR_LENGTH);
     }
     
     ByteArrayOutputStream tmpBout = bout;
     if(!isAscending) {
       // we need to accumulate the bytes in a temp array in order to negate
       // them before writing them to the final array
       tmpBout = new ByteArrayOutputStream();
     }
     
     // now, convert each character to a "code" of one or more bytes
     List<ExtraCodes> unprintableCodes = null;
     List<ExtraCodes> internationalCodes = null;
     int charOffset = 0;
     for(int i = 0; i < str.length(); ++i) {
       char c = str.charAt(i);
       Character cKey = c;
 
       byte[] bytes = CODES.get(cKey);
       if(bytes != null) {
         // simple case, write the codes we found
         tmpBout.write(bytes);
         ++charOffset;
         continue;
       }
 
       bytes = UNPRINTABLE_CODES.get(cKey);
       if(bytes != null) {
         // we do not write anything to tmpBout
         if(bytes.length > 0) {
           if(unprintableCodes == null) {
             unprintableCodes = new LinkedList<ExtraCodes>();
           }
           
           // keep track of the extra codes for later
           unprintableCodes.add(new ExtraCodes(charOffset, bytes));
         }
 
         // note, we do _not_ increment the charOffset for unprintable chars
         continue;
       }
 
       InternationalCodes inatCodes = INTERNATIONAL_CODES.get(cKey);
       if(inatCodes != null) {
 
         // we write the "inline" portion of the international codes
         // immediately, and queue the extra codes for later
         tmpBout.write(inatCodes._inlineCodes);
 
         if(internationalCodes == null) {
           internationalCodes = new LinkedList<ExtraCodes>();
         }
 
         // keep track of the extra codes for later
         internationalCodes.add(new ExtraCodes(charOffset,
                                               inatCodes._extraCodes));
 
         ++charOffset;
         continue;
       }
 
       // bummer, out of luck
       throw new IOException("unmapped string index value " + c);
     }
 
     // write end text flag
     tmpBout.write(END_TEXT);
 
     boolean hasExtraText = ((unprintableCodes != null) ||
                             (internationalCodes != null));
     if(hasExtraText) {
 
       // we write all the international extra bytes first
       if(internationalCodes != null) {
 
         // we write a placeholder char for each non-international char before
         // the extra chars for the international char
         charOffset = 0;
         Iterator<ExtraCodes> iter = internationalCodes.iterator();
         while(iter.hasNext()) {
           ExtraCodes extraCodes = iter.next();
           while(charOffset < extraCodes._charOffset) {
             tmpBout.write(INTERNATIONAL_EXTRA_PLACEHOLDER);
             ++charOffset;
           }
           tmpBout.write(extraCodes._extraCodes);
           ++charOffset;
         }
       }
 
       // then we write all the unprintable extra bytes
       if(unprintableCodes != null) {
 
         // write a single prefix for all unprintable chars
         tmpBout.write(UNPRINTABLE_COMMON_PREFIX);
         
         // we write a whacky combo of bytes for each unprintable char which
         // includes a funky offset and extra char itself
         Iterator<ExtraCodes> iter = unprintableCodes.iterator();
         while(iter.hasNext()) {
           ExtraCodes extraCodes = iter.next();
           int offset =
             (UNPRINTABLE_COUNT_START +
              (UNPRINTABLE_COUNT_MULTIPLIER * extraCodes._charOffset))
             | UNPRINTABLE_OFFSET_FLAGS;
 
           // write offset as big-endian short
           tmpBout.write((offset >> 8) & 0xFF);
           tmpBout.write(offset & 0xFF);
           
           tmpBout.write(UNPRINTABLE_MIDFIX);
           tmpBout.write(extraCodes._extraCodes);
         }
       }
 
     }
 
     // handle descending order by inverting the bytes
     if(!isAscending) {
 
       // we actually write the end byte before flipping the bytes, and write
       // another one after flipping
       tmpBout.write(END_EXTRA_TEXT);
       
       // we actually wrote into a temporary array so that we can invert the
       // bytes before writing them to the final array
       bout.write(flipBytes(tmpBout.toByteArray()));
 
     }
 
     // write end extra text
     bout.write(END_EXTRA_TEXT);    
   }
 
   /**
    * Creates one of the special index entries.
    */
   private static Entry createSpecialEntry(RowId rowId) {
     return new Entry((byte[])null, rowId);
   }
 
   /**
    * Constructs a ColumnDescriptor of the relevant type for the given Column.
    */
   private ColumnDescriptor newColumnDescriptor(Column col, byte flags)
     throws IOException
   {
     switch(col.getType()) {
     case TEXT:
     case MEMO:
       return new TextColumnDescriptor(col, flags);
     case INT:
     case LONG:
     case MONEY:
       return new IntegerColumnDescriptor(col, flags);
     case FLOAT:
     case DOUBLE:
     case SHORT_DATE_TIME:
       return new FloatingPointColumnDescriptor(col, flags);
     case NUMERIC:
       return new FixedPointColumnDescriptor(col, flags);
     case BYTE:
       return new ByteColumnDescriptor(col, flags);
     case BOOLEAN:
       return new BooleanColumnDescriptor(col, flags);
 
     default:
       // FIXME we can't modify this index at this point in time
       setReadOnly();
       return new ReadOnlyColumnDescriptor(col, flags);
     }
   }
 
   /**
    * Returns the EntryType based on the given entry info.
    */
   private static EntryType determineEntryType(byte[] entryBytes, RowId rowId)
   {
     if(entryBytes != null) {
       return ((rowId.getType() == RowId.Type.NORMAL) ?
               EntryType.NORMAL :
               ((rowId.getType() == RowId.Type.ALWAYS_FIRST) ?
                EntryType.FIRST_VALID : EntryType.LAST_VALID));
     } else if(!rowId.isValid()) {
       // this is a "special" entry (first/last)
       return ((rowId.getType() == RowId.Type.ALWAYS_FIRST) ?
               EntryType.ALWAYS_FIRST : EntryType.ALWAYS_LAST);
     }
     throw new IllegalArgumentException("Values was null for valid entry");
   }
 
   /**
    * Returns the maximum amount of entry data which can be encoded on any
    * index page.
    */
   private static int calcMaxPageEntrySize(JetFormat format)
   {
     // the max data we can fit on a page is the min of the space on the page
     // vs the number of bytes which can be encoded in the entry mask
     int pageDataSize = (format.PAGE_SIZE -
                         (format.OFFSET_INDEX_ENTRY_MASK +
                          format.SIZE_INDEX_ENTRY_MASK));
     int entryMaskSize = (format.SIZE_INDEX_ENTRY_MASK * 8);
     return Math.min(pageDataSize, entryMaskSize);
   }
   
   /**
    * Information about the columns in an index.  Also encodes new index
    * values.
    */
   public static abstract class ColumnDescriptor
   {
     private final Column _column;
     private final byte _flags;
 
     private ColumnDescriptor(Column column, byte flags)
       throws IOException
     {
       _column = column;
       _flags = flags;
     }
 
     public Column getColumn() {
       return _column;
     }
 
     public byte getFlags() {
       return _flags;
     }
 
     public boolean isAscending() {
       return((getFlags() & ASCENDING_COLUMN_FLAG) != 0);
     }
     
     public int getColumnIndex() {
       return getColumn().getColumnIndex();
     }
     
     public String getName() {
       return getColumn().getName();
     }
 
     protected boolean isNullValue(Object value) {
       return (value == null);
     }
     
     protected final void writeValue(Object value, ByteArrayOutputStream bout)
       throws IOException
     {
       if(isNullValue(value)) {
         // write null value
         bout.write(getNullEntryFlag(isAscending()));
         return;
       }
       
       // write the start flag
       bout.write(getStartEntryFlag(isAscending()));
       // write the rest of the value
       writeNonNullValue(value, bout);
     }
 
     protected abstract void writeNonNullValue(
         Object value, ByteArrayOutputStream bout)
       throws IOException; 
     
     @Override
     public String toString() {
       return "ColumnDescriptor " + getColumn() + "\nflags: " + getFlags();
     }
   }
 
   /**
    * ColumnDescriptor for integer based columns.
    */
   private static final class IntegerColumnDescriptor extends ColumnDescriptor
   {
     private IntegerColumnDescriptor(Column column, byte flags)
       throws IOException
     {
       super(column, flags);
     }
     
     @Override
     protected void writeNonNullValue(
         Object value, ByteArrayOutputStream bout)
       throws IOException
     {
       byte[] valueBytes = encodeNumberColumnValue(value, getColumn());
       
       // bit twiddling rules:
       // - isAsc  => flipFirstBit
       // - !isAsc => flipFirstBit, flipBytes
       
       flipFirstBitInByte(valueBytes, 0);
       if(!isAscending()) {
         flipBytes(valueBytes);
       }
       
       bout.write(valueBytes);
     }    
   }
   
   /**
    *