/*
    *   Licensed to the Apache Software Foundation (ASF) under one
    *   or more contributor license agreements.  See the NOTICE file
    *   distributed with this work for additional information
    *   regarding copyright ownership.  The ASF licenses this file
    *   to you under the Apache License, Version 2.0 (the
    *   "License"); you may not use this file except in compliance
    *   with the License.  You may obtain a copy of the License at
    *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   *   Unless required by applicable law or agreed to in writing,
   *   software distributed under the License is distributed on an
   *   "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   *   KIND, either express or implied.  See the License for the
   *   specific language governing permissions and limitations
   *   under the License.
   *
   */
  
  package org.apache.directory.mavibot.btree;
  
  
  import java.io.File;
  import java.io.FileInputStream;
  import java.io.FileNotFoundException;
  import java.io.FileOutputStream;
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Comparator;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.TreeMap;
  import java.util.TreeSet;
  
  import org.apache.directory.mavibot.btree.comparator.IntComparator;
  import org.apache.directory.mavibot.btree.exception.EndOfFileExceededException;
  import org.apache.directory.mavibot.btree.exception.KeyNotFoundException;
  import org.apache.directory.mavibot.btree.serializer.IntSerializer;
  
  
  /**
   * A class used to bulk load a BTree. It will allow the load of N elements in 
   * a given BTree without to have to inject one by one, saving a lot of time.
   * The second advantage is that the btree will be dense (the leaves will be
   * complete, except the last one).
   * This class can also be used to compact a BTree.
   *
   * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
   */
  public class BulkLoader<K, V>
  {
      private BulkLoader()
      {
      };
  
      static enum LevelEnum
      {
          LEAF,
          NODE
      }
  
      /**
       * A private class used to store the temporary sorted file. It's used by
       * the bulkLoader
       */
      private static class SortedFile
      {
          /** the file that contains the values */
          private File file;
  
          /** The number of stored values */
          private int nbValues;
  
  
          /** A constructor for this class */
          /*No Qualifier*/SortedFile( File file, int nbValues )
          {
              this.file = file;
              this.nbValues = nbValues;
          }
      }
  
  
      /**
       * Process the data, and creates files to store them sorted if necessary, or store them
       * TODO readElements.
       *
       * @param btree
       * @param iterator
       * @param sortedFiles
       * @param tuples
       * @param chunkSize
       * @return
      * @throws IOException
      */
     private static <K, V> int readElements( BTree<K, V> btree, Iterator<Tuple<K, V>> iterator, List<File> sortedFiles,
         List<Tuple<K, V>> tuples, int chunkSize ) throws IOException
     {
         int nbRead = 0;
         int nbIteration = 0;
         int nbElems = 0;
         boolean inMemory = true;
         Set<K> keys = new HashSet<K>();
 
         while ( true )
         {
             nbIteration++;
             tuples.clear();
             keys.clear();
 
             // Read up to chukSize elements
             while ( iterator.hasNext() && ( nbRead < chunkSize ) )
             {
                 Tuple<K, V> tuple = iterator.next();
                 tuples.add( tuple );
 
                 if ( !keys.contains( tuple.getKey() ) )
                 {
                     keys.add( tuple.getKey() );
                     nbRead++;
                 }
             }
 
             if ( nbRead < chunkSize )
             {
                 if ( nbIteration != 1 )
                 {
                     // Flush the sorted data on disk and exit
                     inMemory = false;
 
                     sortedFiles.add( flushToDisk( nbIteration, tuples, btree ) );
                 }
 
                 // Update the number of read elements
                 nbElems += nbRead;
 
                 break;
             }
             else
             {
                 if ( !iterator.hasNext() )
                 {
                     // special case : we have exactly chunkSize elements in the incoming data
                     if ( nbIteration > 1 )
                     {
                         // Flush the sorted data on disk and exit
                         inMemory = false;
                         sortedFiles.add( flushToDisk( nbIteration, tuples, btree ) );
                     }
 
                     // We have read all the data in one round trip, let's get out, no need
                     // to store the data on disk
 
                     // Update the number of read elements
                     nbElems += nbRead;
 
                     break;
                 }
 
                 // We have read chunkSize elements, we have to sort them on disk
                 nbElems += nbRead;
                 nbRead = 0;
                 sortedFiles.add( flushToDisk( nbIteration, tuples, btree ) );
             }
         }
 
         if ( !inMemory )
         {
             tuples.clear();
         }
 
         return nbElems;
     }
 
 
     /**
      * Read all the sorted files, and inject them into one single big file containing all the 
      * sorted and merged elements.
      * @throws IOException 
      */
     private static <K, V> Tuple<Iterator<Tuple<K, Set<V>>>, SortedFile> processFiles( BTree<K, V> btree,
         Iterator<Tuple<K, Set<V>>> dataIterator ) throws IOException
     {
         File file = File.createTempFile( "sortedUnique", "data" );
         file.deleteOnExit();
         FileOutputStream fos = new FileOutputStream( file );
 
         // Number of read elements
         int nbReads = 0;
 
         // Flush the tuples on disk
         while ( dataIterator.hasNext() )
         {
             nbReads++;
 
             // grab a tuple
             Tuple<K, Set<V>> tuple = dataIterator.next();
 
             // Serialize the key
             byte[] bytesKey = btree.getKeySerializer().serialize( tuple.key );
             fos.write( IntSerializer.serialize( bytesKey.length ) );
             fos.write( bytesKey );
 
             // Serialize the number of values
             int nbValues = tuple.getValue().size();
             fos.write( IntSerializer.serialize( nbValues ) );
 
             // Serialize the values
             for ( V value : tuple.getValue() )
             {
                 byte[] bytesValue = btree.getValueSerializer().serialize( value );
 
                 // Serialize the value
                 fos.write( IntSerializer.serialize( bytesValue.length ) );
                 fos.write( bytesValue );
             }
         }
 
         fos.flush();
         fos.close();
 
         FileInputStream fis = new FileInputStream( file );
         Iterator<Tuple<K, Set<V>>> uniqueIterator = createUniqueFileIterator( btree, fis );
         SortedFile sortedFile = new SortedFile( file, nbReads );
 
         Tuple<Iterator<Tuple<K, Set<V>>>, SortedFile> result = new Tuple<Iterator<Tuple<K, Set<V>>>, SortedFile>(
             uniqueIterator, sortedFile );
 
         return result;
     }
 
 
     /**
      * Bulk Load data into a persisted BTree
      *
      * @param btree The persisted BTree in which we want to load the data
      * @param iterator The iterator over the data to bulkload
      * @param chunkSize The number of elements we may store in memory at each iteration
      * @throws IOException If there is a problem while processing the data
      */
     public static <K, V> BTree<K, V> load( BTree<K, V> btree, Iterator<Tuple<K, V>> iterator, int chunkSize )
         throws IOException
     {
         if ( btree == null )
         {
             throw new RuntimeException( "Invalid BTree : it's null" );
         }
 
         if ( iterator == null )
         {
             // Nothing to do...
             return null;
         }
 
         // Iterate through the elements by chunk
         boolean inMemory = true;
 
         // The list of files we will use to store the sorted chunks
         List<File> sortedFiles = new ArrayList<File>();
 
         // An array of chukSize tuple max
         List<Tuple<K, V>> tuples = new ArrayList<Tuple<K, V>>( chunkSize );
 
         // Now, start to read all the tuples to sort them. We may use intermediate files
         // for that purpose if we hit the threshold.
         int nbElems = readElements( btree, iterator, sortedFiles, tuples, chunkSize );
 
         // If the tuple list is empty, we have to process the load based on files, not in memory
         if ( nbElems > 0 )
         {
             inMemory = tuples.size() > 0;
         }
 
         // Now that we have processed all the data, we can start storing them in the btree
         Iterator<Tuple<K, Set<V>>> dataIterator = null;
         FileInputStream[] streams = null;
         BTree<K, V> resultBTree = null;
 
         if ( inMemory )
         {
             // Here, we have all the data in memory, no need to merge files
             // We will build a simple iterator over the data
             dataIterator = createTupleIterator( btree, tuples );
             resultBTree = bulkLoad( btree, dataIterator, nbElems );
         }
         else
         {
             // We first have to build an iterator over the files
             int nbFiles = sortedFiles.size();
             streams = new FileInputStream[nbFiles];
 
             for ( int i = 0; i < nbFiles; i++ )
             {
                 streams[i] = new FileInputStream( sortedFiles.get( i ) );
             }
 
             dataIterator = createIterator( btree, streams );
 
             // Process the files, and construct one single file with an iterator
             Tuple<Iterator<Tuple<K, Set<V>>>, SortedFile> result = processFiles( btree, dataIterator );
             resultBTree = bulkLoad( btree, result.key, result.value.nbValues );
             result.value.file.delete();
         }
 
         // Ok, we have an iterator over sorted elements, we can now load them in the 
         // target btree.
         // Now, close the FileInputStream, and delete them if we have some
         if ( !inMemory )
         {
             int nbFiles = sortedFiles.size();
 
             for ( int i = 0; i < nbFiles; i++ )
             {
                 streams[i].close();
                 sortedFiles.get( i ).delete();
             }
         }
 
         return resultBTree;
     }
 
 
     /**
      * Creates a node leaf LevelInfo based on the number of elements in the lower level. We can store
      * up to PageSize + 1 references to pages in a node.
      */
     /* no qualifier*/static <K, V> LevelInfo<K, V> computeLevel( BTree<K, V> btree, int nbElems, LevelEnum levelType )
     {
         int pageSize = btree.getPageSize();
         int incrementNode = 0;
 
         if ( levelType == LevelEnum.NODE )
         {
             incrementNode = 1;
         }
 
         LevelInfo<K, V> level = new LevelInfo<K, V>();
         level.setType( ( levelType == LevelEnum.NODE ) );
         level.setNbElems( nbElems );
         level.setNbPages( nbElems / ( pageSize + incrementNode ) );
         level.setLevelNumber( 0 );
         level.setNbAddedElems( 0 );
         level.setCurrentPos( 0 );
 
         // Create the first level page
         if ( nbElems <= pageSize + incrementNode )
         {
             if ( nbElems % ( pageSize + incrementNode ) != 0 )
             {
                 level.setNbPages( 1 );
             }
 
             level.setNbElemsLimit( nbElems );
 
             if ( level.isNode() )
             {
                 level.setCurrentPage( BTreeFactory.createNode( btree, 0L, nbElems - 1 ) );
             }
             else
             {
                 level.setCurrentPage( BTreeFactory.createLeaf( btree, 0L, nbElems ) );
             }
         }
         else
         {
             int remaining = nbElems % ( pageSize + incrementNode );
 
             if ( remaining == 0 )
             {
                 level.setNbElemsLimit( nbElems );
 
                 if ( level.isNode() )
                 {
                     level.setCurrentPage( BTreeFactory.createNode( btree, 0L, pageSize ) );
                 }
                 else
                 {
                     level.setCurrentPage( BTreeFactory.createLeaf( btree, 0L, pageSize ) );
                 }
             }
             else
             {
                 level.incNbPages();
 
                 if ( remaining < ( pageSize / 2 ) + incrementNode )
                 {
                     level.setNbElemsLimit( nbElems - remaining - ( pageSize + incrementNode ) );
 
                     if ( level.getNbElemsLimit() > 0 )
                     {
                         if ( level.isNode() )
                         {
                             level.setCurrentPage( BTreeFactory.createNode( btree, 0L, pageSize ) );
                         }
                         else
                         {
                             level.setCurrentPage( BTreeFactory.createLeaf( btree, 0L, pageSize ) );
                         }
                     }
                     else
                     {
                         if ( level.isNode() )
                         {
                             level
                                 .setCurrentPage( BTreeFactory.createNode( btree, 0L, ( pageSize / 2 ) + remaining - 1 ) );
                         }
                         else
                         {
                             level.setCurrentPage( BTreeFactory.createLeaf( btree, 0L, ( pageSize / 2 ) + remaining ) );
                         }
                     }
                 }
                 else
                 {
                     level.setNbElemsLimit( nbElems - remaining );
 
                     if ( level.isNode() )
                     {
                         level.setCurrentPage( BTreeFactory.createNode( btree, 0L, pageSize ) );
                     }
                     else
                     {
                         level.setCurrentPage( BTreeFactory.createLeaf( btree, 0L, pageSize ) );
                     }
                 }
             }
         }
 
         return level;
     }
 
 
     /**
      * Compute the number of pages necessary to store all the elements per level. The resulting list is
      * reversed ( ie the leaves are on the left, the root page on the right.
      */
     /* No Qualifier */static <K, V> List<LevelInfo<K, V>> computeLevels( BTree<K, V> btree, int nbElems )
     {
         List<LevelInfo<K, V>> levelList = new ArrayList<LevelInfo<K, V>>();
 
         // Compute the leaves info
         LevelInfo<K, V> leafLevel = computeLevel( btree, nbElems, LevelEnum.LEAF );
 
         levelList.add( leafLevel );
         int nbPages = leafLevel.getNbPages();
         int levelNumber = 1;
 
         while ( nbPages > 1 )
         {
             // Compute the Nodes info
             LevelInfo<K, V> nodeLevel = computeLevel( btree, nbPages, LevelEnum.NODE );
             nodeLevel.setLevelNumber( levelNumber++ );
             levelList.add( nodeLevel );
             nbPages = nodeLevel.getNbPages();
         }
 
         return levelList;
     }
 
 
     /**
      * Inject a tuple into a leaf
      */
     private static <K, V> void injectInLeaf( BTree<K, V> btree, Tuple<K, Set<V>> tuple, LevelInfo<K, V> leafLevel )
     {
         PersistedLeaf<K, V> leaf = ( PersistedLeaf<K, V> ) leafLevel.getCurrentPage();
 
         KeyHolder<K> keyHolder = new PersistedKeyHolder<K>( btree.getKeySerializer(), tuple.getKey() );
         leaf.setKey( leafLevel.getCurrentPos(), keyHolder );
 
         if ( btree.getType() != BTreeTypeEnum.PERSISTED_SUB )
         {
             ValueHolder<V> valueHolder = new PersistedValueHolder<V>( btree, ( V[] ) tuple.getValue().toArray() );
             leaf.setValue( leafLevel.getCurrentPos(), valueHolder );
         }
 
         leafLevel.incCurrentPos();
     }
 
 
     private static <K, V> int computeNbElemsLeaf( BTree<K, V> btree, LevelInfo<K, V> levelInfo )
     {
         int pageSize = btree.getPageSize();
         int remaining = levelInfo.getNbElems() - levelInfo.getNbAddedElems();
 
         if ( remaining < pageSize )
         {
             return remaining;
         }
         else if ( remaining == pageSize )
         {
             return pageSize;
         }
         else if ( remaining > levelInfo.getNbElems() - levelInfo.getNbElemsLimit() )
         {
             return pageSize;
         }
         else
         {
             return remaining - pageSize / 2;
         }
     }
 
 
     /**
      * Compute the number of nodes necessary to store all the elements.
      */
     /* No qualifier */int computeNbElemsNode( BTree<K, V> btree, LevelInfo<K, V> levelInfo )
     {
         int pageSize = btree.getPageSize();
         int remaining = levelInfo.getNbElems() - levelInfo.getNbAddedElems();
 
         if ( remaining < pageSize + 1 )
         {
             return remaining;
         }
         else if ( remaining == pageSize + 1 )
         {
             return pageSize + 1;
         }
         else if ( remaining > levelInfo.getNbElems() - levelInfo.getNbElemsLimit() )
         {
             return pageSize + 1;
         }
         else
         {
             return remaining - pageSize / 2;
         }
     }
 
 
     /**
      * Inject a page reference into the root page.
      */
     private static <K, V> void injectInRoot( BTree<K, V> btree, Page<K, V> page, PageHolder<K, V> pageHolder,
         LevelInfo<K, V> level ) throws IOException
     {
         PersistedNode<K, V> node = ( PersistedNode<K, V> ) level.getCurrentPage();
 
         if ( ( level.getCurrentPos() == 0 ) && ( node.getPage( 0 ) == null ) )
         {
             node.setPageHolder( 0, pageHolder );
             level.incNbAddedElems();
         }
         else
         {
             // Inject the pageHolder and the page leftmost key
             node.setPageHolder( level.getCurrentPos() + 1, pageHolder );
             KeyHolder<K> keyHolder = new PersistedKeyHolder<K>( btree.getKeySerializer(), page.getLeftMostKey() );
             node.setKey( level.getCurrentPos(), keyHolder );
             level.incCurrentPos();
             level.incNbAddedElems();
 
             // Check that we haven't added the last element. If so,
             // we have to write the page on disk and update the btree
             if ( level.getNbAddedElems() == level.getNbElems() )
             {
                 PageHolder<K, V> rootHolder = ( ( PersistedBTree<K, V> ) btree ).getRecordManager().writePage(
                     btree, node, 0L );
                 ( ( PersistedBTree<K, V> ) btree ).setRootPage( rootHolder.getValue() );
             }
         }
 
         return;
     }
 
 
     /**
      * Inject a page reference into a Node. This method will recurse if needed.
      */
     private static <K, V> void injectInNode( BTree<K, V> btree, Page<K, V> page, List<LevelInfo<K, V>> levels,
         int levelIndex )
         throws IOException
     {
         int pageSize = btree.getPageSize();
         LevelInfo<K, V> level = levels.get( levelIndex );
         PersistedNode<K, V> node = ( PersistedNode<K, V> ) level.getCurrentPage();
 
         // We first have to write the page on disk
         PageHolder<K, V> pageHolder = ( ( PersistedBTree<K, V> ) btree ).getRecordManager().writePage( btree, page, 0L );
 
         // First deal with a node that has less than PageSize elements at this level.
         // It will become the root node.
         if ( level.getNbElems() <= pageSize + 1 )
         {
             injectInRoot( btree, page, pageHolder, level );
 
             return;
         }
 
         // Now, we have some parent node. We process the 3 different use case :
         // o Full node before the limit
         // o Node over the limit but with at least N/2 elements
         // o Node over the limit but with elements spread into 2 nodes
         if ( level.getNbAddedElems() < level.getNbElemsLimit() )
         {
             // Ok, we haven't yet reached the incomplete pages (if any).
             // Let's add the page reference into the node
             // There is one special case : when we are adding the very first page 
             // reference into a node. In this case, we don't store the key
             if ( ( level.getCurrentPos() == 0 ) && ( node.getKey( 0 ) == null ) )
             {
                 node.setPageHolder( 0, pageHolder );
             }
             else
             {
                 // Inject the pageHolder and the page leftmost key
                 node.setPageHolder( level.getCurrentPos(), pageHolder );
                 KeyHolder<K> keyHolder = new PersistedKeyHolder<K>( btree.getKeySerializer(), page.getLeftMostKey() );
                 node.setKey( level.getCurrentPos() - 1, keyHolder );
             }
 
             // Now, increment this level nb of added elements
             level.incCurrentPos();
             level.incNbAddedElems();
 
             // Check that we haven't added the last element. If so,
             // we have to write the page on disk and update the parent's node
             if ( level.getNbAddedElems() == level.getNbElems() )
             {
                 //PageHolder<K, V> rootHolder = ( ( PersistedBTree<K, V> ) btree ).getRecordManager().writePage(
                 //    btree, node, 0L );
                 //( ( PersistedBTree<K, V> ) btree ).setRootPage( rootHolder.getValue() );
                 injectInNode( btree, node, levels, levelIndex + 1 );
 
                 return;
             }
             else
             {
                 // Check that we haven't completed the current node, and that this is not the root node.
                 if ( ( level.getCurrentPos() == pageSize + 1 ) && ( level.getLevelNumber() < levels.size() - 1 ) )
                 {
                     // yes. We have to write the node on disk, update its parent
                     // and create a new current node
                     injectInNode( btree, node, levels, levelIndex + 1 );
 
                     // The page is full, we have to create a new one, with a size depending on the remaining elements
                     if ( level.getNbAddedElems() < level.getNbElemsLimit() )
                     {
                         // We haven't reached the limit, create a new full node
                         level.setCurrentPage( BTreeFactory.createNode( btree, 0L, pageSize ) );
                     }
                     else if ( level.getNbElems() - level.getNbAddedElems() <= pageSize )
                     {
                         level.setCurrentPage( BTreeFactory.createNode( btree, 0L,
                             level.getNbElems() - level.getNbAddedElems() - 1 ) );
                     }
                     else
                     {
                         level.setCurrentPage( BTreeFactory.createNode( btree, 0L, ( level.getNbElems() - 1 )
                             - ( level.getNbAddedElems() + 1 ) - pageSize / 2 ) );
                     }
 
                     level.setCurrentPos( 0 );
                 }
             }
 
             return;
         }
         else
         {
             // We are dealing with the last page or the last two pages 
             // We can have either one single pages which can contain up to pageSize-1 elements
             // or with two pages, the first one containing ( nbElems - limit ) - pageSize/2 elements
             // and the second one will contain pageSize/2 elements. 
             if ( level.getNbElems() - level.getNbElemsLimit() > pageSize )
             {
                 // As the remaining elements are above a page size, they will be spread across
                 // two pages. We have two cases here, depending on the page we are filling
                 if ( level.getNbElems() - level.getNbAddedElems() <= pageSize / 2 + 1 )
                 {
                     // As we have less than PageSize/2 elements to write, we are on the second page
                     if ( ( level.getCurrentPos() == 0 ) && ( node.getKey( 0 ) == null ) )
                     {
                         node.setPageHolder( 0, pageHolder );
                     }
                     else
                     {
                         // Inject the pageHolder and the page leftmost key
                         node.setPageHolder( level.getCurrentPos(), pageHolder );
                         KeyHolder<K> keyHolder = new PersistedKeyHolder<K>( btree.getKeySerializer(),
                             page.getLeftMostKey() );
                         node.setKey( level.getCurrentPos() - 1, keyHolder );
                     }
 
                     // Now, increment this level nb of added elements
                     level.incCurrentPos();
                     level.incNbAddedElems();
 
                     // Check if we are done with the page
                     if ( level.getNbAddedElems() == level.getNbElems() )
                     {
                         // Yes, we have to update the parent
                         injectInNode( btree, node, levels, levelIndex + 1 );
                     }
                 }
                 else
                 {
                     // This is the first page 
                     if ( ( level.getCurrentPos() == 0 ) && ( node.getKey( 0 ) == null ) )
                     {
                         // First element of the page
                         node.setPageHolder( 0, pageHolder );
                     }
                     else
                     {
                         // Any other following elements
                         // Inject the pageHolder and the page leftmost key
                         node.setPageHolder( level.getCurrentPos(), pageHolder );
                         KeyHolder<K> keyHolder = new PersistedKeyHolder<K>( btree.getKeySerializer(),
                             page.getLeftMostKey() );
                         node.setKey( level.getCurrentPos() - 1, keyHolder );
                     }
 
                     // Now, increment this level nb of added elements
                     level.incCurrentPos();
                     level.incNbAddedElems();
 
                     // Check if we are done with the page
                     if ( level.getCurrentPos() == node.getNbElems() + 1 )
                     {
                         // Yes, we have to update the parent
                         injectInNode( btree, node, levels, levelIndex + 1 );
 
                         // An create a new one
                         level.setCurrentPage( BTreeFactory.createNode( btree, 0L, pageSize / 2 ) );
                         level.setCurrentPos( 0 );
                     }
                 }
             }
             else
             {
                 // Two cases : we don't have anything else to write, or this is a single page
                 if ( level.getNbAddedElems() == level.getNbElems() )
                 {
                     // We are done with the page
                     injectInNode( btree, node, levels, levelIndex + 1 );
                 }
                 else
                 {
                     // We have some more elements to add in  the page
                     // This is the first page 
                     if ( ( level.getCurrentPos() == 0 ) && ( node.getKey( 0 ) == null ) )
                     {
                         // First element of the page
                         node.setPageHolder( 0, pageHolder );
                     }
                     else
                     {
                         // Any other following elements
                         // Inject the pageHolder and the page leftmost key
                         node.setPageHolder( level.getCurrentPos(), pageHolder );
                         KeyHolder<K> keyHolder = new PersistedKeyHolder<K>( btree.getKeySerializer(),
                             page.getLeftMostKey() );
                         node.setKey( level.getCurrentPos() - 1, keyHolder );
                     }
 
                     // Now, increment this level nb of added elements
                     level.incCurrentPos();
                     level.incNbAddedElems();
 
                     // Check if we are done with the page
                     if ( level.getCurrentPos() == node.getNbElems() + 1 )
                     {
                         // Yes, we have to update the parent
                         injectInNode( btree, node, levels, levelIndex + 1 );
 
                         // An create a new one
                         level.setCurrentPage( BTreeFactory.createNode( btree, 0L, pageSize / 2 ) );
                         level.setCurrentPos( 0 );
                     }
                 }
 
                 return;
             }
         }
     }
 
 
     private static <K, V> BTree<K, V> bulkLoadSinglePage( BTree<K, V> btree, Iterator<Tuple<K, Set<V>>> dataIterator,
         int nbElems ) throws IOException
     {
         // Use the root page
         Page<K, V> rootPage = btree.getRootPage();
 
         // Initialize the root page
         ( ( AbstractPage<K, V> ) rootPage ).setNbElems( nbElems );
         KeyHolder<K>[] keys = new KeyHolder[nbElems];
         ValueHolder<V>[] values = new ValueHolder[nbElems];
 
         switch ( btree.getType() )
         {
             case IN_MEMORY:
                 ( ( InMemoryLeaf<K, V> ) rootPage ).values = values;
                 break;
 
             default:
                 ( ( PersistedLeaf<K, V> ) rootPage ).values = values;
         }
 
         // We first have to inject data into the page
         int pos = 0;
 
         while ( dataIterator.hasNext() )
         {
             Tuple<K, Set<V>> tuple = dataIterator.next();
 
             // Store the current element in the rootPage
             KeyHolder<K> keyHolder = new PersistedKeyHolder<K>( btree.getKeySerializer(), tuple.getKey() );
             keys[pos] = keyHolder;
 
             switch ( btree.getType() )
             {
                 case IN_MEMORY:
                     ValueHolder<V> valueHolder = new InMemoryValueHolder<V>( btree, ( V[] ) tuple.getValue()
                         .toArray() );
                     ( ( InMemoryLeaf<K, V> ) rootPage ).values[pos] = valueHolder;
                     break;
 
                 default:
                     valueHolder = new PersistedValueHolder<V>( btree, ( V[] ) tuple.getValue()
                         .toArray() );
                     ( ( PersistedLeaf<K, V> ) rootPage ).values[pos] = valueHolder;
             }
 
             pos++;
         }
 
         // Update the rootPage
         ( ( AbstractPage<K, V> ) rootPage ).setKeys( keys );
 
         // Update the btree with the nb of added elements, and write it$
         BTreeHeader<K, V> btreeHeader = ( ( AbstractBTree<K, V> ) btree ).getBtreeHeader();
         btreeHeader.setNbElems( nbElems );
 
         return btree;
     }
 
 
     /**
      * Construct the target BTree from the sorted data. We will use the nb of elements
      * to determinate the structure of the BTree, as it must be balanced
      */
     private static <K, V> BTree<K, V> bulkLoad( BTree<K, V> btree, Iterator<Tuple<K, Set<V>>> dataIterator, int nbElems )
         throws IOException
     {
         int pageSize = btree.getPageSize();
 
         // Special case : we can store all the element sin a single page
         if ( nbElems <= pageSize )
         {
             return bulkLoadSinglePage( btree, dataIterator, nbElems );
         }
 
         // Ok, we will need more than one page to store the elements, which
         // means we also will need more than one level.
         // First, compute the needed number of levels.
         List<LevelInfo<K, V>> levels = computeLevels( btree, nbElems );
 
         // Now, let's fill the levels
         LevelInfo<K, V> leafLevel = levels.get( 0 );
 
         while ( dataIterator.hasNext() )
         {
             // let's fill page up to the point all the complete pages have been filled
             if ( leafLevel.getNbAddedElems() < leafLevel.getNbElemsLimit() )
             {
                 // grab a tuple
                 Tuple<K, Set<V>> tuple = dataIterator.next();
 
                 injectInLeaf( btree, tuple, leafLevel );
                 leafLevel.incNbAddedElems();
 
                 // The page is completed, update the parent's node and create a new current page
                 if ( leafLevel.getCurrentPos() == pageSize )
                 {
                     injectInNode( btree, leafLevel.getCurrentPage(), levels, 1 );
 
                     // The page is full, we have to create a new one
                     leafLevel.setCurrentPage( BTreeFactory
                         .createLeaf( btree, 0L, computeNbElemsLeaf( btree, leafLevel ) ) );
                     leafLevel.setCurrentPos( 0 );
                 }
             }
             else
             {
                 // We have to deal with uncompleted pages now (if we have any)
                 if ( leafLevel.getNbAddedElems() == nbElems )
                 {
                     // First use case : we have injected all the elements in the btree : get out
                     break;
                 }
 
                 if ( nbElems - leafLevel.getNbElemsLimit() > pageSize )
                 {
                     // Second use case : the number of elements after the limit does not
                     // fit in a page, that means we have to split it into
                     // two pages
 
                     // First page will contain nbElems - leafLevel.nbElemsLimit - PageSize/2 elements
                     int nbToAdd = nbElems - leafLevel.getNbElemsLimit() - pageSize / 2;
 
                     while ( nbToAdd > 0 )
                     {
                         // grab a tuple
                         Tuple<K, Set<V>> tuple = dataIterator.next();
 
                         injectInLeaf( btree, tuple, leafLevel );
                         leafLevel.incNbAddedElems();
                         nbToAdd--;
                     }
 
                     // Now inject the page into the node
                     Page<K, V> currentPage = leafLevel.getCurrentPage();
                     injectInNode( btree, currentPage, levels, 1 );
 
                     // Create a new page for the remaining elements
                     nbToAdd = pageSize / 2;
                     leafLevel.setCurrentPage( BTreeFactory.createLeaf( btree, 0L, nbToAdd ) );
                     leafLevel.setCurrentPos( 0 );
 
                     while ( nbToAdd > 0 )
                     {
                         // grab a tuple
                         Tuple<K, Set<V>> tuple = dataIterator.next();
 
                         injectInLeaf( btree, tuple, leafLevel );
                         leafLevel.incNbAddedElems();
                         nbToAdd--;
                     }
 
                     // And update the parent node
                     Page<K, V> levelCurrentPage = leafLevel.getCurrentPage();
                     injectInNode( btree, levelCurrentPage, levels, 1 );
 
                     // We are done
                     break;
                 }
                 else
                 {
                     // Third use case : we can push all the elements in the last page.
                     // Let's do it
                     int nbToAdd = nbElems - leafLevel.getNbElemsLimit();
 
                     while ( nbToAdd > 0 )
                     {
                         // grab a tuple
                         Tuple<K, Set<V>> tuple = dataIterator.next();
 
                         injectInLeaf( btree, tuple, leafLevel );
                         leafLevel.incNbAddedElems();
                         nbToAdd--;
                     }
 
                     // Now inject the page into the node
                     injectInNode( btree, leafLevel.getCurrentPage(), levels, 1 );
 
                     // and we are done
                     break;
                 }
             }
         }
 
         // Update the btree with the nb of added elements, and write it$
         BTreeHeader<K, V> btreeHeader = ( ( AbstractBTree<K, V> ) btree ).getBtreeHeader();
         btreeHeader.setNbElems( nbElems );
 
         return btree;
     }
 
 
     /**
      * Flush a list of tuples to disk after having sorted them. In the process, we may have to gather the values
      * for the tuples having the same keys.
      * @throws IOException 
      */
     private static <K, V> File flushToDisk( int fileNb, List<Tuple<K, V>> tuples, BTree<K, V> btree )
         throws IOException
     {
         // Sort the tuples. 
         Tuple<K, Set<V>>[] sortedTuples = sort( btree, tuples );
 
         File file = File.createTempFile( "sorted", Integer.toString( fileNb ) );
         file.deleteOnExit();
         FileOutputStream fos = new FileOutputStream( file );
 
         // Flush the tuples on disk
         for ( Tuple<K, Set<V>> tuple : sortedTuples )
         {
             // Serialize the key
             byte[] bytesKey = btree.getKeySerializer().serialize( tuple.key );
             fos.write( IntSerializer.serialize( bytesKey.length ) );
             fos.write( bytesKey );
 
             // Serialize the number of values
             int nbValues = tuple.getValue().size();
             fos.write( IntSerializer.serialize( nbValues ) );
 
             // Serialize the values
             for ( V value : tuple.getValue() )
             {
                 byte[] bytesValue = btree.getValueSerializer().serialize( value );
 
                 // Serialize the value
                 fos.write( IntSerializer.serialize( bytesValue.length ) );
                 fos.write( bytesValue );
             }
         }
 
         fos.flush();
         fos.close();
 
         return file;
     }
 
 
     /**
      * Sort a list of tuples, eliminating the duplicate keys and storing the values in a set when we 
      * have a duplicate key
      */
     private static <K, V> Tuple<K, Set<V>>[] sort( BTree<K, V> btree, List<Tuple<K, V>> tuples )
     {
         Comparator<Tuple<K, Set<V>>> tupleComparator = new TupleComparator( btree.getKeyComparator(), btree
             .getValueComparator() );
 
         // Sort the list
         Tuple<K, V>[] tuplesArray = ( Tuple<K, V>[] ) tuples.toArray( new Tuple[]
             {} );
 
         // First, eliminate the equals keys. We use a map for that
         Map<K, Set<V>> mapTuples = new HashMap<K, Set<V>>();
 
         for ( Tuple<K, V> tuple : tuplesArray )
         {
             // Is the key present in the map ?
             Set<V> foundSet = mapTuples.get( tuple.key );
 
             if ( foundSet != null )
             {
                 // We already have had such a key, add the value to the existing key
                 foundSet.add( tuple.value );
             }
             else
             {
                 // No such key present in the map : create a new set to store the values,
                 // and add it in the map associated with the new key
                 Set<V> set = new TreeSet<V>();
                 set.add( tuple.value );
                 mapTuples.put( tuple.key, set );
             }
         }
 
         // Now, sort the map, by extracting all the key/values from the map
         int size = mapTuples.size();
         Tuple<K, Set<V>>[] sortedTuples = new Tuple[size];
         int pos = 0;
 
         // We create an array containing all the elements
         for ( Map.Entry<K, Set<V>> entry : mapTuples.entrySet() )
         {
             sortedTuples[pos] = new Tuple<K, Set<V>>();
             sortedTuples[pos].key = entry.getKey();
             sortedTuples[pos].value = entry.getValue();
             pos++;
         }
 
         // And we sort the array
         Arrays.sort( sortedTuples, tupleComparator );
 
         return sortedTuples;
     }
 
 
     /**
      * Build an iterator over an array of sorted tuples, in memory
      */
     private static <K, V> Iterator<Tuple<K, Set<V>>> createTupleIterator( BTree<K, V> btree, List<Tuple<K, V>> tuples )
     {
         final Tuple<K, Set<V>>[] sortedTuples = sort( btree, tuples );
 
         Iterator<Tuple<K, Set<V>>> tupleIterator = new Iterator<Tuple<K, Set<V>>>()
         {
             private int pos = 0;
 
 
             @Override
             public Tuple<K, Set<V>> next()
             {
                 // Return the current tuple, if any
                 if ( pos < sortedTuples.length )
                 {
                     Tuple<K, Set<V>> tuple = sortedTuples[pos];
                     pos++;
 
                     return tuple;
                 }
 
                 return null;
             }
 
 
             @Override
             public boolean hasNext()
             {
                 return pos < sortedTuples.length;
             }
 
 
             @Override
             public void remove()
             {
             }
         };
 
         return tupleIterator;
     }
 
 
     private static <K, V> Tuple<K, Set<V>> fetchTuple( BTree<K, V> btree, FileInputStream fis )
     {
         try
         {
             if ( fis.available() == 0 )
             {
                 return null;
             }
 
             Tuple<K, Set<V>> tuple = new Tuple<K, Set<V>>();
             tuple.value = new TreeSet<V>();
 
             byte[] intBytes = new byte[4];
 
             // Read the key length
             fis.read( intBytes );
             int keyLength = IntSerializer.deserialize( intBytes );
 
             // Read the key
             byte[] keyBytes = new byte[keyLength];
             fis.read( keyBytes );
             K key = btree.getKeySerializer().fromBytes( keyBytes );
             tuple.key = key;
 
             // get the number of values
             fis.read( intBytes );
             int nbValues = IntSerializer.deserialize( intBytes );
 
             // Serialize the values
             for ( int i = 0; i < nbValues; i++ )
             {
                 // Read the value length
                 fis.read( intBytes );
                 int valueLength = IntSerializer.deserialize( intBytes );
 
                 // Read the value
                 byte[] valueBytes = new byte[valueLength];
                 fis.read( valueBytes );
                 V value = btree.getValueSerializer().fromBytes( valueBytes );
                 tuple.value.add( value );
             }
 
             return tuple;
         }
         catch ( IOException ioe )
         {
             return null;
         }
     }
 
 
     /**
      * Build an iterator over an array of sorted tuples, from files on the disk
      * @throws FileNotFoundException 
      */
     private static <K, V> Iterator<Tuple<K, Set<V>>> createIterator( final BTree<K, V> btree,
         final FileInputStream[] streams )
         throws FileNotFoundException
     {
         // The number of files we have to read from
         final int nbFiles = streams.length;
 
         // We will read only one element at a time from each file
         final Tuple<K, Set<V>>[] readTuples = new Tuple[nbFiles];
         final TreeMap<Tuple<K, Integer>, Set<V>> candidates =
             new TreeMap<Tuple<K, Integer>, Set<V>>(
                 new TupleComparator<K, Integer>( btree.getKeyComparator(), IntComparator.INSTANCE ) );
 
         // Read the tuple from each files
         for ( int i = 0; i < nbFiles; i++ )
         {
             while ( true )
             {
                 readTuples[i] = fetchTuple( btree, streams[i] );
 
                 if ( readTuples[i] != null )
                 {
                     Tuple<K, Integer> candidate = new Tuple<K, Integer>( readTuples[i].key, i, btree.getKeySerializer()
                         .getComparator() );
 
                     if ( !candidates.containsKey( candidate ) )
                     {
                         candidates.put( candidate, readTuples[i].getValue() );
                         break;
                     }
                     else
                     {
                         // We have to merge the pulled tuple with the existing one, and read one more tuple
                         Set<V> oldValues = candidates.get( candidate );
                         oldValues.addAll( readTuples[i].getValue() );
                     }
                 }
                 else
                 {
                     break;
                 }
             }
         }
 
         Iterator<Tuple<K, Set<V>>> tupleIterator = new Iterator<Tuple<K, Set<V>>>()
         {
             @Override
             public Tuple<K, Set<V>> next()
             {
                 // Get the first candidate
                 Tuple<K, Integer> tupleCandidate = candidates.firstKey();
 
                 // Remove it from the set
                 candidates.remove( tupleCandidate );
 
                 // Get the the next tuple from the stream we just got the tuple from
                 Tuple<K, Set<V>> tuple = readTuples[tupleCandidate.value];
 
                 // fetch the next tuple from the file we just read teh candidate from 
                 while ( true )
                 {
                     // fetch it from the disk and store it into its reader
                     readTuples[tupleCandidate.value] = fetchTuple( btree, streams[tupleCandidate.value] );
 
                     if ( readTuples[tupleCandidate.value] == null )
                     {
                         // No more tuple for this file
                         break;
                     }
 
                     if ( readTuples[tupleCandidate.value] != null )
                     {
                         // And store it into the candidate set
                         Set<V> oldValues = candidates.get( readTuples[tupleCandidate.value] );
 
                         if ( oldValues != null )
                         {
                             // We already have another element with the same key, merge them
                             oldValues.addAll( readTuples[tupleCandidate.value].value );
                         }
                         else
                         {
                             Tuple<K, Integer> newTuple = new Tuple<K, Integer>( readTuples[tupleCandidate.value].key,
                                 tupleCandidate.value );
                             candidates.put( newTuple, readTuples[tupleCandidate.value].getValue() );
 
                             // and exit the loop
                             break;
                         }
                     }
                 }
 
                 // We can now return the found value
                 return tuple;
             }
 
 
             @Override
             public boolean hasNext()
             {
                 // Check that we have at least one element to read
                 return !candidates.isEmpty();
             }
 
 
             @Override
             public void remove()
             {
             }
 
         };
 
         return tupleIterator;
     }
 
 
     /**
      * Build an iterator over an array of sorted tuples, from files on the disk
      * @throws FileNotFoundException 
      */
     private static <K, V> Iterator<Tuple<K, Set<V>>> createUniqueFileIterator( final BTree<K, V> btree,
         final FileInputStream stream )
         throws FileNotFoundException
     {
         Iterator<Tuple<K, Set<V>>> tupleIterator = new Iterator<Tuple<K, Set<V>>>()
         {
             boolean hasNext = true;
 
 
             @Override
             public Tuple<K, Set<V>> next()
             {
                 // Get the tuple from the stream
                 Tuple<K, Set<V>> tuple = fetchTuple( btree, stream );
 
                 // We can now return the found value
                 return tuple;
             }
 
 
             @Override
             public boolean hasNext()
             {
                 // Check that we have at least one element to read
                 try
                 {
                     return stream.available() > 0;
                 }
                 catch ( IOException e )
                 {
                     return false;
                 }
             }
 
 
             @Override
             public void remove()
             {
             }
 
         };
 
         return tupleIterator;
     }
 
 
     /**
      * Compact a given persisted BTree, making it dense. All the values will be stored 
      * in newly created pages, each one of them containing as much elements
      * as it's size.
      * </br>
      * The RecordManager will be stopped and restarted, do not use this method
      * on a running BTree.
      *
      * @param recordManager The associated recordManager
      * @param btree The BTree to compact
      */
     public static void compact( RecordManager recordManager, BTree<?, ?> btree )
     {
 
     }
 
 
     /**
      * Compact a given in-memory BTree, making it dense. All the values will be stored 
      * in newly created pages, each one of them containing as much elements
      * as it's size.
      * </br>
      *
      * @param btree The BTree to compact
      * @throws KeyNotFoundException 
      * @throws IOException 
      */
     public static BTree<?, ?> compact( BTree<?, ?> btree ) throws IOException, KeyNotFoundException
     {
         // First, create a new BTree which will contain all the elements
         InMemoryBTreeConfiguration configuration = new InMemoryBTreeConfiguration();
         configuration.setName( btree.getName() );
         configuration.setPageSize( btree.getPageSize() );
         configuration.setKeySerializer( btree.getKeySerializer() );
         configuration.setValueSerializer( btree.getValueSerializer() );
         configuration.setAllowDuplicates( btree.isAllowDuplicates() );
         configuration.setReadTimeOut( btree.getReadTimeOut() );
         configuration.setWriteBufferSize( btree.getWriteBufferSize() );
 
         File file = ( ( InMemoryBTree ) btree ).getFile();
 
         if ( file != null )
         {
             configuration.setFilePath( file.getPath() );
         }
 
         // Create a new Btree Builder
         InMemoryBTreeBuilder btreeBuilder = new InMemoryBTreeBuilder( configuration );
 
         // Create a cursor over the existing btree
         final TupleCursor cursor = btree.browse();
 
         // Create an iterator that will iterate the existing btree
         Iterator<Tuple> tupleItr = new Iterator<Tuple>()
         {
             @Override
             public Tuple next()
             {
                 try
                 {
                     return cursor.next();
                 }
                 catch ( EndOfFileExceededException e )
                 {
                     return null;
                 }
                 catch ( IOException e )
                 {
                     return null;
                 }
             }
 
 
             @Override
             public boolean hasNext()
             {
                 try
                 {
                     return cursor.hasNext();
                 }
                 catch ( EndOfFileExceededException e )
                 {
                     return false;
                 }
                 catch ( IOException e )
                 {
                     return false;
                 }
             }
 
 
             @Override
             public void remove()
             {
             }
         };
 
         // And finally, compact the btree
         return btreeBuilder.build( tupleItr );
     }
 }