/*
   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.IOException;
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.Iterator;
  import java.util.LinkedHashMap;
  import java.util.List;
  import java.util.Map;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  /**
   * A single database table
   * <p>
   * Is not thread-safe.
   * 
   * @author Tim McCune
   */
  public class Table
    implements Iterable<Map<String, Object>>
  {
    
    private static final Log LOG = LogFactory.getLog(Table.class);
  
    private static final short OFFSET_MASK = (short)0x1FFF;
  
    private static final short DELETED_ROW_MASK = (short)0x8000;
    
    private static final short OVERFLOW_ROW_MASK = (short)0x4000;
  
    /** Table type code for system tables */
    public static final byte TYPE_SYSTEM = 0x53;
    /** Table type code for user tables */
    public static final byte TYPE_USER = 0x4e;
  
    /** comparator which sorts variable length columns vased on their index into
        the variable length offset table */
    private static final Comparator<Column> VAR_LEN_COLUMN_COMPARATOR =
      new Comparator<Column>() {
        public int compare(Column c1, Column c2) {
          return ((c1.getVarLenTableIndex() < c2.getVarLenTableIndex()) ? -1 :
                  ((c1.getVarLenTableIndex() > c2.getVarLenTableIndex()) ? 1 :
                   0));
        }
      };
  
    /** owning database */
    private final Database _database;
    /** Type of the table (either TYPE_SYSTEM or TYPE_USER) */
    private byte _tableType;
    /** Number of indexes on the table */
    private int _indexCount;
    /** Number of index slots for the table */
    private int _indexSlotCount;
    /** Number of rows in the table */
    private int _rowCount;
    /** last auto number for the table */
    private int _lastAutoNumber;
    /** page number of the definition of this table */
    private final int _tableDefPageNumber;
    /** max Number of columns in the table (includes previous deletions) */
    private short _maxColumnCount;
    /** max Number of variable columns in the table */
    private short _maxVarColumnCount;
    /** List of columns in this table, ordered by column number */
    private List<Column> _columns = new ArrayList<Column>();
   /** List of variable length columns in this table, ordered by offset */
   private List<Column> _varColumns = new ArrayList<Column>();
   /** List of indexes on this table */
   private List<Index> _indexes = new ArrayList<Index>();
   /** Table name as stored in Database */
   private final String _name;
   /** Usage map of pages that this table owns */
   private UsageMap _ownedPages;
   /** Usage map of pages that this table owns with free space on them */
   private UsageMap _freeSpacePages;
   /** modification count for the table, keeps row-states up-to-date */
   private int _modCount;
   /** page buffer used to update data pages when adding rows */
   private final TempPageHolder _addRowBufferH =
     TempPageHolder.newHolder(TempBufferHolder.Type.SOFT);
   /** page buffer used to update the table def page */
   private final TempPageHolder _tableDefBufferH =
     TempPageHolder.newHolder(TempBufferHolder.Type.SOFT);
   /** buffer used to writing single rows of data */
   private final TempBufferHolder _singleRowBufferH =
     TempBufferHolder.newHolder(TempBufferHolder.Type.SOFT, true);
   /** "buffer" used to writing multi rows of data (will create new buffer on
       every call) */
   private final TempBufferHolder _multiRowBufferH =
     TempBufferHolder.newHolder(TempBufferHolder.Type.NONE, true);
   /** for now, "big index support" is optional */
   private final boolean _useBigIndex;
   
   /** common cursor for iterating through the table, kept here for historic
       reasons */
   private Cursor _cursor;
   
   /**
    * Only used by unit tests
    */
   Table(boolean testing, List<Column> columns) throws IOException {
     if(!testing) {
       throw new IllegalArgumentException();
     }
     _database = null;
     _tableDefPageNumber = PageChannel.INVALID_PAGE_NUMBER;
     _name = null;
     _useBigIndex = false;
     setColumns(columns);
   }
   
   /**
    * @param database database which owns this table
    * @param tableBuffer Buffer to read the table with
    * @param pageNumber Page number of the table definition
    * @param name Table name
    * @param useBigIndex whether or not "big index support" should be enabled
    *                    for the table
    */
   protected Table(Database database, ByteBuffer tableBuffer,
                   int pageNumber, String name, boolean useBigIndex)
   throws IOException
   {
     _database = database;
     _tableDefPageNumber = pageNumber;
     _name = name;
     _useBigIndex = useBigIndex; 
     int nextPage = tableBuffer.getInt(getFormat().OFFSET_NEXT_TABLE_DEF_PAGE);
     ByteBuffer nextPageBuffer = null;
     while (nextPage != 0) {
       if (nextPageBuffer == null) {
         nextPageBuffer = getPageChannel().createPageBuffer();
       }
       getPageChannel().readPage(nextPageBuffer, nextPage);
       nextPage = nextPageBuffer.getInt(getFormat().OFFSET_NEXT_TABLE_DEF_PAGE);
       ByteBuffer newBuffer = getPageChannel().createBuffer(
           tableBuffer.capacity() + getFormat().PAGE_SIZE - 8);
       newBuffer.put(tableBuffer);
       newBuffer.put(nextPageBuffer.array(), 8, getFormat().PAGE_SIZE - 8);
       tableBuffer = newBuffer;
       tableBuffer.flip();
     }
     readTableDefinition(tableBuffer);
     tableBuffer = null;
 
     // setup common cursor
     _cursor = Cursor.createCursor(this);
   }
 
   /**
    * @return The name of the table
    */
   public String getName() {
     return _name;
   }
 
   public boolean doUseBigIndex() {
     return _useBigIndex;
   }
   
   public int getMaxColumnCount() {
     return _maxColumnCount;
   }
   
   public int getColumnCount() {
     return _columns.size();
   }
   
   public Database getDatabase() {
     return _database;
   }
   
   public JetFormat getFormat() {
     return getDatabase().getFormat();
   }
 
   public PageChannel getPageChannel() {
     return getDatabase().getPageChannel();
   }
 
   protected int getTableDefPageNumber() {
     return _tableDefPageNumber;
   }
 
   public RowState createRowState() {
     return new RowState(TempBufferHolder.Type.HARD);
   }
 
   protected UsageMap.PageCursor getOwnedPagesCursor() {
     return _ownedPages.cursor();
   }
   
   /**
    * @return All of the columns in this table (unmodifiable List)
    */
   public List<Column> getColumns() {
     return Collections.unmodifiableList(_columns);
   }
 
   /**
    * @return the column with the given name
    */
   public Column getColumn(String name) {
     for(Column column : _columns) {
       if(column.getName().equals(name)) {
         return column;
       }
     }
     throw new IllegalArgumentException("Column with name " + name +
                                        " does not exist in this table");
   }
     
   /**
    * Only called by unit tests
    */
   private void setColumns(List<Column> columns) {
     _columns = columns;
     int colIdx = 0;
     int varLenIdx = 0;
     int fixedOffset = 0;
     for(Column col : _columns) {
       col.setColumnNumber((short)colIdx);
       col.setColumnIndex(colIdx++);
       if(col.isVariableLength()) {
         col.setVarLenTableIndex(varLenIdx++);
         _varColumns.add(col);
       } else {
         col.setFixedDataOffset(fixedOffset);
         fixedOffset += col.getType().getFixedSize();
       }
     }
     _maxColumnCount = (short)_columns.size();
     _maxVarColumnCount = (short)_varColumns.size();
   }
   
   /**
    * @return All of the Indexes on this table (unmodifiable List)
    */
   public List<Index> getIndexes() {
     return Collections.unmodifiableList(_indexes);
   }
 
   /**
    * @return the index with the given name
    */
   public Index getIndex(String name) {
     for(Index index : _indexes) {
       if(index.getName().equals(name)) {
         return index;
       }
     }
     throw new IllegalArgumentException("Index with name " + name +
                                        " does not exist on this table");
   }
     
   /**
    * Only called by unit tests
    */
   int getIndexSlotCount() {
     return _indexSlotCount;
   }
 
   /**
    * After calling this method, getNextRow will return the first row in the
    * table
    */
   public void reset() {
     _cursor.reset();
   }
   
   /**
    * Delete the current row (retrieved by a call to {@link #getNextRow()}).
    */
   public void deleteCurrentRow() throws IOException {
     _cursor.deleteCurrentRow();
   }
 
   /**
    * Delete the row on which the given rowState is currently positioned.
    */
   public void deleteRow(RowState rowState, RowId rowId) throws IOException {
     requireValidRowId(rowId);
     
     // ensure that the relevant row state is up-to-date
     ByteBuffer rowBuffer = positionAtRowHeader(rowState, rowId);
 
     requireNonDeletedRow(rowState, rowId);
     
     // delete flag always gets set in the "header" row (even if data is on
     // overflow row)
     int pageNumber = rowState.getHeaderRowId().getPageNumber();
     int rowNumber = rowState.getHeaderRowId().getRowNumber();
 
     // use any read rowValues to help update the indexes
     Object[] rowValues = (!_indexes.isEmpty() ?
                           rowState.getRowValues() : null);
     
     int rowIndex = getRowStartOffset(rowNumber, getFormat());
     rowBuffer.putShort(rowIndex, (short)(rowBuffer.getShort(rowIndex)
                                       | DELETED_ROW_MASK | OVERFLOW_ROW_MASK));
     writeDataPage(rowBuffer, pageNumber);
 
     // update the indexes
     for(Index index : _indexes) {
       index.deleteRow(rowValues, rowId);
     }
     
     // make sure table def gets updated
     updateTableDefinition(-1);
   }
   
   /**
    * @return The next row in this table (Column name -> Column value)
    */
   public Map<String, Object> getNextRow() throws IOException {
     return getNextRow(null);
   }
 
   /**
    * @param columnNames Only column names in this collection will be returned
    * @return The next row in this table (Column name -> Column value)
    */
   public Map<String, Object> getNextRow(Collection<String> columnNames) 
     throws IOException
   {
     return _cursor.getNextRow(columnNames);
   }
   
   /**
    * Reads a single column from the given row.
    */
   public Object getRowValue(RowState rowState, RowId rowId, Column column)
     throws IOException
   {
     if(this != column.getTable()) {
       throw new IllegalArgumentException(
           "Given column " + column + " is not from this table");
     }
     requireValidRowId(rowId);
     
     // position at correct row
     ByteBuffer rowBuffer = positionAtRowData(rowState, rowId);
     requireNonDeletedRow(rowState, rowId);
     
     Object value = getRowColumn(rowBuffer, getRowNullMask(rowBuffer), column);
 
     // cache the row values in order to be able to update the index on row
     // deletion.  note, most of the returned values are immutable, except
     // for binary data (returned as byte[]), but binary data shouldn't be
     // indexed anyway.
     rowState.setRowValue(column.getColumnIndex(), value);
 
     return value;
   }
 
   /**
    * Reads some columns from the given row.
    * @param columnNames Only column names in this collection will be returned
    */
   public Map<String, Object> getRow(
       RowState rowState, RowId rowId, Collection<String> columnNames)
     throws IOException
   {
     requireValidRowId(rowId);
 
     // position at correct row
     ByteBuffer rowBuffer = positionAtRowData(rowState, rowId);
     requireNonDeletedRow(rowState, rowId);
 
     return getRow(rowState, rowBuffer, getRowNullMask(rowBuffer), _columns,
                   columnNames);
   }
 
   /**
    * Reads the row data from the given row buffer.  Leaves limit unchanged.
    * Saves parsed row values to the given rowState.
    */
   private static Map<String, Object> getRow(
       RowState rowState,
       ByteBuffer rowBuffer,
       NullMask nullMask,
       Collection<Column> columns,
       Collection<String> columnNames)
     throws IOException
   {
     Map<String, Object> rtn = new LinkedHashMap<String, Object>(
         columns.size());
     for(Column column : columns) {
 
       if((columnNames == null) || (columnNames.contains(column.getName()))) {
         // Add the value to the row data
         Object value = getRowColumn(rowBuffer, nullMask, column);
         rtn.put(column.getName(), value);
 
         // cache the row values in order to be able to update the index on row
         // deletion.  note, most of the returned values are immutable, except
         // for binary data (returned as byte[]), but binary data shouldn't be
         // indexed anyway.
         rowState.setRowValue(column.getColumnIndex(), value);
       }
     }
     return rtn;
   }
   
   /**
    * Reads the column data from the given row buffer.  Leaves limit unchanged.
    */
   private static Object getRowColumn(ByteBuffer rowBuffer,
                                      NullMask nullMask,
                                      Column column)
     throws IOException
   {
     boolean isNull = nullMask.isNull(column);
     if(column.getType() == DataType.BOOLEAN) {
       return Boolean.valueOf(!isNull);  //Boolean values are stored in the null mask
     } else if(isNull) {
       // well, that's easy!
       return null;
     }
 
     // reset position to row start
     rowBuffer.reset();
     
     // locate the column data bytes
     int rowStart = rowBuffer.position();
     int colDataPos = 0;
     int colDataLen = 0;
     if(!column.isVariableLength()) {
 
       // read fixed length value (non-boolean at this point)
       int dataStart = rowStart + 2;
       colDataPos = dataStart + column.getFixedDataOffset();
       colDataLen = column.getType().getFixedSize();
       
     } else {
 
       // read var length value
       int varColumnOffsetPos =
         (rowBuffer.limit() - nullMask.byteSize() - 4) -
         (column.getVarLenTableIndex() * 2);
 
       short varDataStart = rowBuffer.getShort(varColumnOffsetPos);
       short varDataEnd = rowBuffer.getShort(varColumnOffsetPos - 2);
       colDataPos = rowStart + varDataStart;
       colDataLen = varDataEnd - varDataStart;
     }
 
     // grab the column data
     byte[] columnData = new byte[colDataLen];
     rowBuffer.position(colDataPos);
     rowBuffer.get(columnData);
 
     // parse the column data
     return column.read(columnData);
   }
 
   /**
    * Reads the null mask from the given row buffer.  Leaves limit unchanged.
    */
   private static NullMask getRowNullMask(ByteBuffer rowBuffer)
     throws IOException
   {
     // reset position to row start
     rowBuffer.reset();
     
     short columnCount = rowBuffer.getShort(); // Number of columns in this row
     
     // read null mask
     NullMask nullMask = new NullMask(columnCount);
     rowBuffer.position(rowBuffer.limit() - nullMask.byteSize());  //Null mask at end
     nullMask.read(rowBuffer);
 
     return nullMask;
   }
 
   /**
    * Sets a new buffer to the correct row header page using the given rowState
    * according to the given rowId.  Deleted state is
    * determined, but overflow row pointers are not followed.
    * 
    * @return a ByteBuffer of the relevant page, or null if row was invalid
    */
   public static ByteBuffer positionAtRowHeader(RowState rowState,
                                                RowId rowId)
     throws IOException
   {
     ByteBuffer rowBuffer = rowState.setHeaderRow(rowId);
 
     if(rowState.isAtHeaderRow()) {
       // this task has already been accomplished
       return rowBuffer;
     }
     
     if(!rowState.isValid()) {
       // this was an invalid page/row
       rowState.setStatus(RowStateStatus.AT_HEADER);
       return null;
     }
 
     // note, we don't use findRowStart here cause we need the unmasked value
     short rowStart = rowBuffer.getShort(
         getRowStartOffset(rowId.getRowNumber(),
                           rowState.getTable().getFormat()));
 
     // check the deleted, overflow flags for the row (the "real" flags are
     // always set on the header row)
     RowStatus rowStatus = RowStatus.NORMAL;
     if(isDeletedRow(rowStart)) {
       rowStatus = RowStatus.DELETED;
     } else if(isOverflowRow(rowStart)) {
       rowStatus = RowStatus.OVERFLOW;
     }
 
     rowState.setRowStatus(rowStatus);
     rowState.setStatus(RowStateStatus.AT_HEADER);
     return rowBuffer;
   }
   
   /**
    * Sets the position and limit in a new buffer using the given rowState
    * according to the given row number and row end, following overflow row
    * pointers as necessary.
    * 
    * @return a ByteBuffer narrowed to the actual row data, or null if row was
    *         invalid or deleted
    */
   public static ByteBuffer positionAtRowData(RowState rowState,
                                              RowId rowId)
     throws IOException
   {
     positionAtRowHeader(rowState, rowId);
     if(!rowState.isValid() || rowState.isDeleted()) {
       // row is invalid or deleted
       rowState.setStatus(RowStateStatus.AT_FINAL);
       return null;
     }
 
     ByteBuffer rowBuffer = rowState.getFinalPage();
     int rowNum = rowState.getFinalRowId().getRowNumber();
     JetFormat format = rowState.getTable().getFormat();
     
     if(rowState.isAtFinalRow()) {
       // we've already found the final row data
       return PageChannel.narrowBuffer(
           rowBuffer,
           findRowStart(rowBuffer, rowNum, format),
           findRowEnd(rowBuffer, rowNum, format));
     }
     
     while(true) {
       
       // note, we don't use findRowStart here cause we need the unmasked value
       short rowStart = rowBuffer.getShort(getRowStartOffset(rowNum, format));
       short rowEnd = findRowEnd(rowBuffer, rowNum, format);
 
       // note, at this point we know the row is not deleted, so ignore any
       // subsequent deleted flags (as overflow rows are always marked deleted
       // anyway)
       boolean overflowRow = isOverflowRow(rowStart);
 
       // now, strip flags from rowStart offset
       rowStart = (short)(rowStart & OFFSET_MASK);
 
       if (overflowRow) {
 
         if((rowEnd - rowStart) < 4) {
           throw new IOException("invalid overflow row info");
         }
       
         // Overflow page.  the "row" data in the current page points to
         // another page/row
         int overflowRowNum = ByteUtil.getUnsignedByte(rowBuffer, rowStart);
         int overflowPageNum = ByteUtil.get3ByteInt(rowBuffer, rowStart + 1);
         rowBuffer = rowState.setOverflowRow(
             new RowId(overflowPageNum, overflowRowNum));
         rowNum = overflowRowNum;
       
       } else {
 
         rowState.setStatus(RowStateStatus.AT_FINAL);
         return PageChannel.narrowBuffer(rowBuffer, rowStart, rowEnd);
       }
     }
   }
 
   
   /**
    * Calls <code>reset</code> on this table and returns an unmodifiable
    * Iterator which will iterate through all the rows of this table.  Use of
    * the Iterator follows the same restrictions as a call to
    * <code>getNextRow</code>.
    * @throws IllegalStateException if an IOException is thrown by one of the
    *         operations, the actual exception will be contained within
    */
   public Iterator<Map<String, Object>> iterator()
   {
     return iterator(null);
   }
   
   /**
    * Calls <code>reset</code> on this table and returns an unmodifiable
    * Iterator which will iterate through all the rows of this table, returning
    * only the given columns.  Use of the Iterator follows the same
    * restrictions as a call to <code>getNextRow</code>.
    * @throws IllegalStateException if an IOException is thrown by one of the
    *         operations, the actual exception will be contained within
    */
   public Iterator<Map<String, Object>> iterator(Collection<String> columnNames)
   {
     reset();
     return _cursor.iterator(columnNames);
   }
 
   /**
    * Writes a new table defined by the given columns to the database.
    * @return the first page of the new table's definition
    */
   public static int writeTableDefinition(
       List<Column> columns, PageChannel pageChannel, JetFormat format)
     throws IOException
   {
     // first, create the usage map page
     int usageMapPageNumber = pageChannel.writeNewPage(
         createUsageMapDefinitionBuffer(pageChannel, format));
 
     // next, determine how big the table def will be (in case it will be more
     // than one page)
     int totalTableDefSize = format.SIZE_TDEF_HEADER +
       (format.SIZE_COLUMN_DEF_BLOCK * columns.size()) +
       format.SIZE_TDEF_TRAILER;
     for(Column col : columns) {
       // we add the number of bytes for the column name and 2 bytes for the
       // length of the column name
       int nameByteLen = (col.getName().length() *
                          JetFormat.TEXT_FIELD_UNIT_SIZE);
       totalTableDefSize += nameByteLen + 2;
     }
     
     // now, create the table definition
     ByteBuffer buffer = pageChannel.createBuffer(Math.max(totalTableDefSize,
                                                           format.PAGE_SIZE));
     writeTableDefinitionHeader(buffer, columns, usageMapPageNumber,
                                totalTableDefSize, format);
     writeColumnDefinitions(buffer, columns, format); 
     
     //End of tabledef
     buffer.put((byte) 0xff);
     buffer.put((byte) 0xff);
 
     // write table buffer to database
     int tdefPageNumber = PageChannel.INVALID_PAGE_NUMBER;
     if(totalTableDefSize <= format.PAGE_SIZE) {
       
       // easy case, fits on one page
       buffer.putShort(format.OFFSET_FREE_SPACE,
                       (short)(buffer.remaining() - 8)); // overwrite page free space
       // Write the tdef page to disk.
       tdefPageNumber = pageChannel.writeNewPage(buffer);
       
     } else {
 
       // need to split across multiple pages
       ByteBuffer partialTdef = pageChannel.createPageBuffer();
       buffer.rewind();
       int nextTdefPageNumber = PageChannel.INVALID_PAGE_NUMBER;
       while(buffer.hasRemaining()) {
 
         // reset for next write
         partialTdef.clear();
         
         if(tdefPageNumber == PageChannel.INVALID_PAGE_NUMBER) {
           
           // this is the first page.  note, the first page already has the
           // page header, so no need to write it here
           tdefPageNumber = pageChannel.allocateNewPage();
           nextTdefPageNumber = tdefPageNumber;
           
         } else {
 
           // write page header
           writeTablePageHeader(partialTdef);
         }
 
         // copy the next page of tdef bytes
         int curTdefPageNumber = nextTdefPageNumber;
         int writeLen = Math.min(partialTdef.remaining(), buffer.remaining());
         partialTdef.put(buffer.array(), buffer.position(), writeLen);
         buffer.position(buffer.position() + writeLen);
 
         if(buffer.hasRemaining()) {
           // need a next page
           nextTdefPageNumber = pageChannel.allocateNewPage();
           partialTdef.putInt(format.OFFSET_NEXT_TABLE_DEF_PAGE,
                              nextTdefPageNumber);
         }
 
         // update page free space
         partialTdef.putShort(format.OFFSET_FREE_SPACE,
                              (short)(partialTdef.remaining() - 8)); // overwrite page free space
 
         // write partial page to disk
         pageChannel.writePage(partialTdef, curTdefPageNumber);
       }
         
     }
        
     return tdefPageNumber;
   }
 
   /**
    * @param buffer Buffer to write to
    * @param columns List of Columns in the table
    */
   private static void writeTableDefinitionHeader(
       ByteBuffer buffer, List<Column> columns,
       int usageMapPageNumber, int totalTableDefSize, JetFormat format)
     throws IOException
   {
     //Start writing the tdef
     writeTablePageHeader(buffer);
     buffer.putInt(totalTableDefSize);  //Length of table def
     buffer.put((byte) 0x59);  //Unknown
     buffer.put((byte) 0x06);  //Unknown
     buffer.putShort((short) 0); //Unknown
     buffer.putInt(0);  //Number of rows
     buffer.putInt(0); //Last Autonumber
     if(countAutoNumberColumns(columns) > 0) {
       buffer.put((byte) 1);
     } else {
       buffer.put((byte) 0);
     }
     for (int i = 0; i < 15; i++) {  //Unknown
       buffer.put((byte) 0);
     }
     buffer.put(Table.TYPE_USER); //Table type
     buffer.putShort((short) columns.size()); //Max columns a row will have
     buffer.putShort(Column.countVariableLength(columns));  //Number of variable columns in table
     buffer.putShort((short) columns.size()); //Number of columns in table
     buffer.putInt(0);  //Number of indexes in table
     buffer.putInt(0);  //Number of indexes in table
     buffer.put((byte) 0); //Usage map row number
     ByteUtil.put3ByteInt(buffer, usageMapPageNumber);  //Usage map page number
     buffer.put((byte) 1); //Free map row number
     ByteUtil.put3ByteInt(buffer, usageMapPageNumber);  //Free map page number
     if (LOG.isDebugEnabled()) {
       int position = buffer.position();
       buffer.rewind();
       LOG.debug("Creating new table def block:\n" + ByteUtil.toHexString(
           buffer, format.SIZE_TDEF_HEADER));
       buffer.position(position);
     }
   }
 
   /**
    * Writes the page header for a table definition page
    * @param buffer Buffer to write to
    */
   private static void writeTablePageHeader(ByteBuffer buffer)
   {
     buffer.put(PageTypes.TABLE_DEF);  //Page type
     buffer.put((byte) 0x01); //Unknown
     buffer.put((byte) 0); //Unknown
     buffer.put((byte) 0); //Unknown
     buffer.putInt(0);  //Next TDEF page pointer
   }
   
   /**
    * @param buffer Buffer to write to
    * @param columns List of Columns to write definitions for
    */
   private static void writeColumnDefinitions(
       ByteBuffer buffer, List<Column> columns, JetFormat format)
     throws IOException
   {
     short columnNumber = (short) 0;
     short fixedOffset = (short) 0;
     short variableOffset = (short) 0;
     // we specifically put the "long variable" values after the normal
     // variable length values so that we have a better chance of fitting it
     // all (because "long variable" values can go in separate pages)
     short longVariableOffset =
       Column.countNonLongVariableLength(columns);
     for (Column col : columns) {
       int position = buffer.position();
       buffer.put(col.getType().getValue());
       buffer.put((byte) 0x59);  //Unknown
       buffer.put((byte) 0x06);  //Unknown
       buffer.putShort((short) 0); //Unknown
       buffer.putShort(columnNumber);  //Column Number
       if (col.isVariableLength()) {
         if(!col.getType().isLongValue()) {
           buffer.putShort(variableOffset++);
         } else {
           buffer.putShort(longVariableOffset++);
         }          
       } else {
         buffer.putShort((short) 0);
       }
       buffer.putShort(columnNumber); //Column Number again
       if(col.getType().getHasScalePrecision()) {
         buffer.put(col.getPrecision());  // numeric precision
         buffer.put(col.getScale());  // numeric scale
       } else {
         buffer.put((byte) 0x00); //unused
         buffer.put((byte) 0x00); //unused
       }
       buffer.putShort((short) 0); //Unknown
       buffer.put(getColumnBitFlags(col)); // misc col flags
       if (col.isCompressedUnicode()) {  //Compressed
         buffer.put((byte) 1);
       } else {
         buffer.put((byte) 0);
       }
       buffer.putInt(0); //Unknown, but always 0.
       //Offset for fixed length columns
       if (col.isVariableLength()) {
         buffer.putShort((short) 0);
       } else {
         buffer.putShort(fixedOffset);
         fixedOffset += col.getType().getFixedSize();
       }
       if(!col.getType().isLongValue()) {
         buffer.putShort(col.getLength()); //Column length
       } else {
         buffer.putShort((short)0x0000); // unused
       }
       columnNumber++;
       if (LOG.isDebugEnabled()) {
         LOG.debug("Creating new column def block\n" + ByteUtil.toHexString(
             buffer, position, format.SIZE_COLUMN_DEF_BLOCK));
       }
     }
     for (Column col : columns) {
       writeName(buffer, col.getName(), format);
     }
   }
 
   /**
    * Writes the given name into the given buffer in the format as expected by
    * {@link #readName}.
    */
   private static void writeName(ByteBuffer buffer, String name,
                                 JetFormat format)
   {
       ByteBuffer encName = Column.encodeUncompressedText(
           name, format);
       buffer.putShort((short) encName.remaining());
       buffer.put(encName);
   }
 
   /**
    * Constructs a byte containing the flags for the given column.
    */
   private static byte getColumnBitFlags(Column col) {
     byte flags = Column.UNKNOWN_FLAG_MASK;
     if(!col.isVariableLength()) {
       flags |= Column.FIXED_LEN_FLAG_MASK;
     }
     if(col.isAutoNumber()) {
       flags |= Column.AUTO_NUMBER_FLAG_MASK;
     }
     return flags;
   }
   
   /**
    * Create the usage map definition page buffer.  The "used pages" map is in
    * row 0, the "pages with free space" map is in row 1.
    */
   private static ByteBuffer createUsageMapDefinitionBuffer(
       PageChannel pageChannel, JetFormat format)
     throws IOException
   {
     int usageMapRowLength = format.OFFSET_USAGE_MAP_START +
       format.USAGE_MAP_TABLE_BYTE_LENGTH;
     int freeSpace = format.PAGE_INITIAL_FREE_SPACE
       - (2 * getRowSpaceUsage(usageMapRowLength, format));
     
     ByteBuffer rtn = pageChannel.createPageBuffer();
     rtn.put(PageTypes.DATA);
     rtn.put((byte) 0x1);  //Unknown
     rtn.putShort((short)freeSpace);  //Free space in page
     rtn.putInt(0); //Table definition
     rtn.putInt(0); //Unknown
     rtn.putShort((short) 2); //Number of records on this page
 
     // write two rows of usage map definitions
     int rowStart = findRowEnd(rtn, 0, format) - usageMapRowLength;
     for(int i = 0; i < 2; ++i) {
       rtn.putShort(getRowStartOffset(i, format), (short)rowStart);
       if(i == 0) {
         // initial "usage pages" map definition
         rtn.put(rowStart, UsageMap.MAP_TYPE_REFERENCE);
       } else {
         // initial "pages with free space" map definition
         rtn.put(rowStart, UsageMap.MAP_TYPE_INLINE);
       }
       rowStart -= usageMapRowLength;
     }
         
     return rtn;
   }
     
   /**
    * Read the table definition
    */
   private void readTableDefinition(ByteBuffer tableBuffer) throws IOException
   {
     if (LOG.isDebugEnabled()) {
       tableBuffer.rewind();
       LOG.debug("Table def block:\n" + ByteUtil.toHexString(tableBuffer,
           getFormat().SIZE_TDEF_HEADER));
     }
     _rowCount = tableBuffer.getInt(getFormat().OFFSET_NUM_ROWS);
     _lastAutoNumber = tableBuffer.getInt(getFormat().OFFSET_NEXT_AUTO_NUMBER);
     _tableType = tableBuffer.get(getFormat().OFFSET_TABLE_TYPE);
     _maxColumnCount = tableBuffer.getShort(getFormat().OFFSET_MAX_COLS);
     _maxVarColumnCount = tableBuffer.getShort(getFormat().OFFSET_NUM_VAR_COLS);
     short columnCount = tableBuffer.getShort(getFormat().OFFSET_NUM_COLS);
     _indexSlotCount = tableBuffer.getInt(getFormat().OFFSET_NUM_INDEX_SLOTS);
     _indexCount = tableBuffer.getInt(getFormat().OFFSET_NUM_INDEXES);
 
     int rowNum = ByteUtil.getUnsignedByte(
         tableBuffer, getFormat().OFFSET_OWNED_PAGES);
     int pageNum = ByteUtil.get3ByteInt(tableBuffer, getFormat().OFFSET_OWNED_PAGES + 1);
     _ownedPages = UsageMap.read(getDatabase(), pageNum, rowNum, false);
     rowNum = ByteUtil.getUnsignedByte(
         tableBuffer, getFormat().OFFSET_FREE_SPACE_PAGES);
     pageNum = ByteUtil.get3ByteInt(tableBuffer, getFormat().OFFSET_FREE_SPACE_PAGES + 1);
     _freeSpacePages = UsageMap.read(getDatabase(), pageNum, rowNum, false);
     
     for (int i = 0; i < _indexCount; i++) {
       int uniqueEntryCountOffset =
         (getFormat().OFFSET_INDEX_DEF_BLOCK +
          (i * getFormat().SIZE_INDEX_DEFINITION) + 4);
       int uniqueEntryCount = tableBuffer.getInt(uniqueEntryCountOffset);
       _indexes.add(createIndex(uniqueEntryCount, uniqueEntryCountOffset));
     }
     
     int colOffset = getFormat().OFFSET_INDEX_DEF_BLOCK +
         _indexCount * getFormat().SIZE_INDEX_DEFINITION;
     for (int i = 0; i < columnCount; i++) {
       Column column = new Column(this, tableBuffer,
           colOffset + (i * getFormat().SIZE_COLUMN_HEADER));
       _columns.add(column);
       if(column.isVariableLength()) {
         // also shove it in the variable columns list, which is ordered
         // differently from the _columns list
         _varColumns.add(column);
       }
     }
     tableBuffer.position(colOffset +
                          (columnCount * getFormat().SIZE_COLUMN_HEADER));
     for (int i = 0; i < columnCount; i++) {
       Column column = _columns.get(i);
       column.setName(readName(tableBuffer));
     }    
     Collections.sort(_columns);
 
     // setup the data index for the columns
     int colIdx = 0;
     for(Column col : _columns) {
       col.setColumnIndex(colIdx++);
     }
 
     // sort variable length columns based on their index into the variable
     // length offset table, because we will write the columns in this order
     Collections.sort(_varColumns, VAR_LEN_COLUMN_COMPARATOR);
     
     int idxOffset = tableBuffer.position();
     tableBuffer.position(idxOffset +
                      (getFormat().OFFSET_INDEX_NUMBER_BLOCK * _indexCount));
 
     // if there are more index slots than indexes, the initial slots are
     // always empty/invalid, so we skip that data
     int firstRealIdx = (_indexSlotCount - _indexCount);
     
     for (int i = 0; i < _indexSlotCount; i++) {
 
       tableBuffer.getInt(); //Forward past Unknown
       tableBuffer.getInt(); //Forward past alternate index number
       int indexNumber = tableBuffer.getInt();
       tableBuffer.position(tableBuffer.position() + 11);
       byte indexType = tableBuffer.get();
       tableBuffer.position(tableBuffer.position() + 4);
 
       if(i < firstRealIdx) {
         // ignore this info
         continue;
       }
 
       Index index = _indexes.get(i - firstRealIdx);
       index.setIndexNumber(indexNumber);
       index.setIndexType(indexType);
     }
 
     // read actual index names
     for (int i = 0; i < _indexSlotCount; i++) {
       if(i < firstRealIdx) {
         // for each empty index slot, there is some weird sort of name, skip
         // it
         skipName(tableBuffer);
         continue;
       }
         
       _indexes.get(i - firstRealIdx)
         .setName(readName(tableBuffer));
     }
     int idxEndOffset = tableBuffer.position();
     
     Collections.sort(_indexes);
 
     // go back to index column info after sorting
     tableBuffer.position(idxOffset);
     for (int i = 0; i < _indexCount; i++) {
       tableBuffer.getInt(); //Forward past Unknown
       _indexes.get(i).read(tableBuffer, _columns);
     }
 
     // reset to end of index info
     tableBuffer.position(idxEndOffset);
   }
 
   /**
    * Creates an index with the given initial info.
    */
   private Index createIndex(int uniqueEntryCount, int uniqueEntryCountOffset)
   {
     return(_useBigIndex ?
            new BigIndex(this, uniqueEntryCount, uniqueEntryCountOffset) :
            new SimpleIndex(this, uniqueEntryCount, uniqueEntryCountOffset));
   }
   
   /**
    * Writes the given page data to the given page number, clears any other
    * relevant buffers.
    */
   private void writeDataPage(ByteBuffer pageBuffer, int pageNumber)
     throws IOException
   {
     // write the page data
     getPageChannel().writePage(pageBuffer, pageNumber);
 
     // possibly invalidate the add row buffer if a different data buffer is
     // being written (e.g. this happens during deleteRow)
     _addRowBufferH.possiblyInvalidate(pageNumber, pageBuffer);
     
     // update modification count so any active RowStates can keep themselves
     // up-to-date
     ++_modCount;
   }
 
   /**
    * Returns a name read from the buffer at the current position.  The
    * expected name format is the name length as a short followed by (length *
    * 2) bytes encoded using the {@link JetFormat#CHARSET}
    */
   private String readName(ByteBuffer buffer) {
     int nameLength = ByteUtil.getUnsignedShort(buffer);
     byte[] nameBytes = new byte[nameLength];
     buffer.get(nameBytes);
     return Column.decodeUncompressedText(nameBytes, getFormat());
   }
   
   /**
    * Skips past a name int the buffer at the current position.  The
    * expected name format is the same as that for {@link #readName}.
    */
   private void skipName(ByteBuffer buffer) {
     int nameLength = ByteUtil.getUnsignedShort(buffer);
     buffer.position(buffer.position() + nameLength);
   }
   
   /**
    * Converts a map of columnName -> columnValue to an array of row values
    * appropriate for a call to {@link #addRow(Object...)}.
    */
   public Object[] asRow(Map<String,Object> rowMap) {
     Object[] row = new Object[_columns.size()];
     if(rowMap == null) {
       return row;
     }
     for(Column col : _columns) {
       row[col.getColumnIndex()] = rowMap.get(col.getName());
     }
     return row;
   }
   
   /**
    * Add a single row to this table and write it to disk
    * <p>
    * Note, if this table has an auto-number column, the value written will be
    * put back into the given row array.
    *
    * @param row row values for a single row.  the row will be modified if
    *            this table contains an auto-number column, otherwise it
    *            will not be modified.
    */
   public void addRow(Object... row) throws IOException {
     addRows(Collections.singletonList(row), _singleRowBufferH);
   }
   
   /**
    * Add multiple rows to this table, only writing to disk after all
    * rows have been written, and every time a data page is filled.  This
    * is much more efficient than calling <code>addRow</code> multiple times.
    * <p>
    * Note, if this table has an auto-number column, the values written will be
    * put back into the given row arrays.
    * 
    * @param rows List of Object[] row values.  the rows will be modified if
    *             this table contains an auto-number column, otherwise they
    *             will not be modified.
    */
   public void addRows(List<? extends Object[]> rows) throws IOException {
     addRows(rows, _multiRowBufferH);
   }
   
   /**
    * Add multiple rows to this table, only writing to disk after all
    * rows have been written, and every time a data page is filled.
    * @param inRows List of Object[] row values
    * @param writeRowBufferH TempBufferHolder used to generate buffers for
    *                        writing the row data
    */
   private void addRows(List<? extends Object[]> inRows,
                        TempBufferHolder writeRowBufferH)
     throws IOException
   {
     // copy the input rows to a modifiable list so we can update the elements
     List<Object[]> rows = new ArrayList<Object[]>(inRows);
     ByteBuffer[] rowData = new ByteBuffer[rows.size()];
     for (int i = 0; i < rows.size(); i++) {
 
       // we need to make sure the row is the right length (fill with null).
       // note, if the row is copied the caller will not be able to access any
       // generated auto-number value, but if they need that info they should
       // use a row array of the right size!
       Object[] row = rows.get(i);
       if(row.length < _columns.size()) {
         row = dupeRow(row, _columns.size());
         // we copied the row, so put the copy back into the rows list
         rows.set(i, row);
       }
 
       // write the row of data to a temporary buffer
       rowData[i] = createRow(row, getFormat().MAX_ROW_SIZE,
                              writeRowBufferH.getPageBuffer(getPageChannel()));
       
       if (rowData[i].limit() > getFormat().MAX_ROW_SIZE) {
         throw new IOException("Row size " + rowData[i].limit() +
                               " is too large");
       }
     }
     
     ByteBuffer dataPage = null;
     int pageNumber = PageChannel.INVALID_PAGE_NUMBER;
 
     // find last data page (Not bothering to check other pages for free
     // space.)
     UsageMap.PageCursor revPageCursor = _ownedPages.cursor();
     revPageCursor.afterLast();
     while(true) {
       int tmpPageNumber = revPageCursor.getPreviousPage();
       if(tmpPageNumber < 0) {
         break;
       }
       dataPage = _addRowBufferH.setPage(getPageChannel(), tmpPageNumber);
       if(dataPage.get() == PageTypes.DATA) {
         // found last data page, only use if actually listed in free space
         // pages
         if(_freeSpacePages.containsPageNumber(tmpPageNumber)) {
           pageNumber = tmpPageNumber;
         }
         break;
       }
     }
 
     if(pageNumber == PageChannel.INVALID_PAGE_NUMBER) {
       // No data pages exist (with free space).  Create a new one.
       dataPage = newDataPage();
       pageNumber = _addRowBufferH.getPageNumber();
     }
     
     for (int i = 0; i < rowData.length; i++) {
       int rowSize = rowData[i].remaining();
       int rowSpaceUsage = getRowSpaceUsage(rowSize, getFormat());
       short freeSpaceInPage = dataPage.getShort(getFormat().OFFSET_FREE_SPACE);
       int rowsOnPage = getRowsOnDataPage(dataPage, getFormat());
       if((freeSpaceInPage < rowSpaceUsage) ||
          (rowsOnPage >= getFormat().MAX_NUM_ROWS_ON_DATA_PAGE)) {
 
         // Last data page is full.  Create a new one.
         writeDataPage(dataPage, pageNumber);
         _freeSpacePages.removePageNumber(pageNumber);
 
         dataPage = newDataPage();
         pageNumber = _addRowBufferH.getPageNumber();
         
         freeSpaceInPage = dataPage.getShort(getFormat().OFFSET_FREE_SPACE);
       }
 
       // write out the row data
       int rowNum = addDataPageRow(dataPage, rowSize, getFormat());
       dataPage.put(rowData[i]);
 
       // update the indexes
       for(Index index : _indexes) {
         index.addRow(rows.get(i), new RowId(pageNumber, rowNum));
       }
     }
     writeDataPage(dataPage, pageNumber);
     
     // Update tdef page
     updateTableDefinition(rows.size());
   }
 
   /**
    * Updates the table definition after rows are modified.
    */
   private void updateTableDefinition(int rowCountInc) throws IOException
   {
     // load table definition
     ByteBuffer tdefPage = _tableDefBufferH.setPage(getPageChannel(),
                                                    _tableDefPageNumber);
     
     // make sure rowcount and autonumber are up-to-date
     _rowCount += rowCountInc;
     tdefPage.putInt(getFormat().OFFSET_NUM_ROWS, _rowCount);
     tdefPage.putInt(getFormat().OFFSET_NEXT_AUTO_NUMBER, _lastAutoNumber);
 
     // write any index changes
     Iterator<Index> indIter = _indexes.iterator();
     for (int i = 0; i < _indexes.size(); i++) {
       Index index = indIter.next();
       // write the unique entry count for the index to the table definition
       // page
       tdefPage.putInt(index.getUniqueEntryCountOffset(),
                       index.getUniqueEntryCount());
       // write the entry page for the index
       index.update();
     }
 
     // write modified table definition
     getPageChannel().writePage(tdefPage, _tableDefPageNumber);
   }
   
   /**
    * Create a new data page
    * @return Page number of the new page
    */
   private ByteBuffer newDataPage() throws IOException {
     if (LOG.isDebugEnabled()) {
       LOG.debug("Creating new data page");
     }
     ByteBuffer dataPage = _addRowBufferH.setNewPage(getPageChannel());
     dataPage.put(PageTypes.DATA); //Page type
     dataPage.put((byte) 1); //Unknown
     dataPage.putShort((short)getFormat().PAGE_INITIAL_FREE_SPACE); //Free space in this page
     dataPage.putInt(_tableDefPageNumber); //Page pointer to table definition
     dataPage.putInt(0); //Unknown
     dataPage.putShort((short)0); //Number of rows on this page
     int pageNumber = _addRowBufferH.getPageNumber();
     getPageChannel().writePage(dataPage, pageNumber);
     _ownedPages.addPageNumber(pageNumber);
     _freeSpacePages.addPageNumber(pageNumber);
     return dataPage;
   }
   
   /**
    * Serialize a row of Objects into a byte buffer.
    * <p>
    * Note, if this table has an auto-number column, the value written will be
    * put back into the given row array.
    * 
    * @param rowArray row data, expected to be correct length for this table
    * @param maxRowSize max size the data can be for this row
    * @param buffer buffer to which to write the row data
    * @return the given buffer, filled with the row data
    */
   ByteBuffer createRow(Object[] rowArray, int maxRowSize, ByteBuffer buffer)
     throws IOException
   {
     buffer.putShort(_maxColumnCount);
     NullMask nullMask = new NullMask(_maxColumnCount);
     
     //Fixed length column data comes first
     int fixedDataStart = buffer.position();
     int fixedDataEnd = fixedDataStart;
     for (Column col : _columns) {
 
       if(!col.isVariableLength()) {
         
         Object rowValue = rowArray[col.getColumnIndex()];
 
         if (col.getType() == DataType.BOOLEAN) {
         
           if(Column.toBooleanValue(rowValue)) {