/* Pepito v1.59  -- file: btree.c --
   Copyright (c) 2000-01 Carlos del Cacho

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2, or (at your option)
   any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; see the file COPYING.  If not, write to
   the Free Software Foundation, Inc., 59 Temple Place - Suite 330,
   Boston, MA 02111-1307, USA.

   Contact the author: guggenjaimito@yahoo.com
*/

#ifndef LINUX
#include <windows.h>
#include <winbase.h>
#include <wincon.h>
#include <io.h>
#include <time.h>
#else
#include <sys/time.h>
#endif

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <assert.h>
#include <math.h>
#include "btree.h"
#include "stack.h"

#define BTREE_VER      0xff00ff00
#define INDEX_FILE              0
#define DATA_FILE               1

#define MAX_DATA_CACHE      50000
#define INPUT_BUFFER_SIZE     512
#define MAX_RUN_BUFFER     100000
#define MERGE_WAYS              4

#define swap(a,b,type) { type _tmp_ = a; a = b; b = _tmp_; }

#ifndef Min
#define Min(a,b)  ((a) <= (b) ? (a) : (b))
#endif

#ifndef Max
#define Max(a,b)  ((a) >= (b) ? (a) : (b))
#endif


#define logn(n,x) (log(x)/log(n))

#define PAGE_SIZE(dbh)  (dbh->fheader.page_size * (dbh->fheader.key_size + \
                         (dbh->fheader.data_size ? 2 : 3 )*sizeof(offset_t)) +\
                         sizeof(offset_t) + sizeof(page_header))

#define hash_fun(offset)  ((offset / PAGE_SIZE(dbh)) & (dbh->cache_pipe_size-1))


//-------------------------------------------------------------------
#include <sys/types.h>
#include <sys/stat.h>

#ifndef LINUX
#define stat   _stat
#endif

//-------------------------------------------------------------------
unsigned long FileSize(char *fname)
{
    struct stat buffer;

    if (stat(fname,&buffer) == -1)
        return 0;

    return buffer.st_size;
}

//-------------------------------------------------------------------
db_handle *CreateDatabase(char *file, int key_size, int data_size, int cache_size)
{
    db_handle *dbh;
    db_page *p;
    FILE *fp;
    char fname[256];
    int i;

    strcpy(fname, file);
    strcat(fname,".ind");

    if (!(fp=fopen(fname,"w+b")))
        return NULL;

    dbh = malloc(sizeof(db_handle));

    if (dbh == NULL)
    {
        fclose(fp);
        return NULL;
    }

    dbh->fheader.version = BTREE_VER;
    dbh->fheader.file_type = INDEX_FILE;
    dbh->fheader.file_size = sizeof(file_header);
    dbh->fheader.key_size = key_size;
    dbh->fheader.data_size = data_size;
    dbh->fheader.nentries = 0;
    dbh->fheader.page_size = (4096 - sizeof(offset_t) - sizeof(page_header)) /
                             (dbh->fheader.key_size + 3*sizeof(offset_t));
                             
    printf("Setting page size to %d entries\n",dbh->fheader.page_size);
    dbh->fheader.root_off = sizeof(file_header);
    dbh->fp = fp;

    InitCache(dbh, cache_size, 4);
    SL_Init(&dbh->data_cache, sizeof(offset_t));

    fseek(fp,0,SEEK_SET);
    fwrite(&(dbh->fheader), sizeof(file_header), 1, fp);
    fflush(fp);

    p = AllocPage(dbh);
    p->info.is_leaf = 1;
    p->info.nentries = 0;
    p->page_off = 0;
    StorePage(dbh,p);
    FreePage(p);

    i = strlen(fname);

    fname[i-3] = 'b';
    fname[i-2] = 'i';
    fname[i-1] = 'n';

    if (!(dbh->fdata = fopen(fname,"w+b")))
        return NULL;

    dbh->data_off = 4;
    fwrite(&dbh->data_off, sizeof(offset_t), 1, dbh->fdata);
    fflush(dbh->fdata);

    return dbh;
}

//-------------------------------------------------------------------
db_handle *OpenDatabase(char *file, int cache_size)
{
    db_handle *dbh;
    FILE *fp;
    char fname[256];
    int i;

    strcpy(fname, file);
    strcat(fname,".ind");

    if (!(fp=fopen(fname,"r+b")))
        return NULL;

    dbh = malloc(sizeof(db_handle));

    if (dbh == NULL)
    {
        fclose(fp);
        return NULL;
    }

    fseek(fp,0,SEEK_SET);

    if (fread(&(dbh->fheader),sizeof(file_header),1,fp) != 1
        || dbh->fheader.version != BTREE_VER)
    {
        free(dbh);
        fclose(fp);
        return NULL;
    }

    dbh->fp = fp;

    InitCache(dbh, cache_size, 4);
    SL_Init(&dbh->data_cache, sizeof(offset_t));

    i = strlen(fname);

    fname[i-3] = 'b';
    fname[i-2] = 'i';
    fname[i-1] = 'n';

    if (!(dbh->fdata = fopen(fname,"r+b")))
        return NULL;

    fread(&dbh->data_off, sizeof(offset_t), 1, dbh->fdata);
    return dbh;
}
//-------------------------------------------------------------------
void CloseDatabase(db_handle *dbh)
{
    CloseCache(dbh);
    DataBufferFlush(dbh);
    SL_DestroyList(&dbh->data_cache);

    fseek(dbh->fp,0,SEEK_SET);
    fwrite(&dbh->fheader, sizeof(file_header), 1, dbh->fp);
    fclose(dbh->fp);
    fseek(dbh->fdata,0,SEEK_SET);
    fwrite(&dbh->data_off, sizeof(offset_t), 1, dbh->fdata);
    fclose(dbh->fdata);
}
//-------------------------------------------------------------------
db_page *AllocPage(db_handle *dbh)
{
    db_page *p = malloc(sizeof(db_page));

    if (p)
    {
       p->keys = malloc((dbh->fheader.page_size+1) * dbh->fheader.key_size);

       if (!p->keys)
       {
           free(p);
           return NULL;
       }

       p->childs = malloc((dbh->fheader.page_size+2) * sizeof(offset_t));

       if (!p->childs)
       {
           free(p->keys);
           free(p);
           return NULL;
       }

       p->data_off = malloc((dbh->fheader.page_size+1) * sizeof(offset_t));

       if (!p->data_off)
       {
           free(p->childs);
           free(p->keys);
           free(p);
           return NULL;
       }

       p->data_size = malloc((dbh->fheader.page_size+1) * sizeof(offset_t));

       if (!p->data_size)
       {
           free(p->data_off);
           free(p->childs);
           free(p->keys);
           free(p);
           return NULL;
       }

    }
        
    return p;
}
//-------------------------------------------------------------------
void FreePage(db_page *p)
{
   free(p->keys);
   free(p->childs);
   free(p->data_off);
   free(p->data_size);
   free(p);
}
//-------------------------------------------------------------------
void LoadPage (db_handle    *dbh,
               db_page      *p,
               uint32       offset)
{
    int key = hash_fun(offset);
    unsigned int pipe;

    assert(offset != 0);

    for(pipe = 0; pipe < dbh->npipes; pipe++)
    {
       if (dbh->cache[pipe][key].page->page_off == offset)
       {
          ClonePage(dbh, p, dbh->cache[pipe][key].page);
          while (pipe > 0)
          {
              swap(dbh->cache[pipe], dbh->cache[pipe-1], cache_entry *);
              pipe--;
          }
          return;
       }
    }

    pipe = dbh->npipes-1;

    if (dbh->cache[pipe][key].dirty)
    {
       WritePageToDisk(dbh,dbh->cache[pipe][key].page);
       dbh->cache[pipe][key].dirty = 0;
    }

    while (pipe > 0)
    {
       swap(dbh->cache[pipe], dbh->cache[pipe-1], cache_entry *);
       pipe--;
    }

    fseek(dbh->fp, offset, SEEK_SET);
    fread(&dbh->cache[0][key].page->info, sizeof(page_header), 1, dbh->fp);
    fread(dbh->cache[0][key].page->keys, dbh->fheader.key_size, 
          dbh->fheader.page_size, dbh->fp);
    fread(dbh->cache[0][key].page->childs, sizeof(offset_t),
          1+dbh->fheader.page_size, dbh->fp);
    fread(dbh->cache[0][key].page->data_off, sizeof(offset_t), 
          dbh->fheader.page_size, dbh->fp);

    if (!dbh->fheader.data_size)
      fread(dbh->cache[0][key].page->data_size, sizeof(offset_t),
            dbh->cache[0][key].page->info.nentries, dbh->fp);

    dbh->cache[0][key].page->page_off = offset;
    ClonePage(dbh, p, dbh->cache[0][key].page);
}
//-----------------------------------------------------------------------
void FillPage (db_handle    *dbh,
               db_page      *p,
               int          key_size,
               void         *keys,
               offset_t     *data_off,
               offset_t     *data_size,
               int          nkeys,
               offset_t     *childs)
{
    unsigned int i = nkeys+1;
    memcpy(p->keys, keys, nkeys*key_size);
    memcpy(p->childs, childs, (nkeys+1)*sizeof(offset_t));
    memcpy(p->data_off, data_off, nkeys*sizeof(offset_t));
    memcpy(p->data_size, data_size, nkeys*sizeof(offset_t));

    do {

        p->childs[++i] = 0;

    } while (i <= dbh->fheader.page_size);

    p->info.nentries = nkeys;
    p->page_off = 0;
}
//-----------------------------------------------------------------------
void ClonePage (db_handle    *dbh,
                db_page      *dst,
                db_page      *src)
{

    dst->info = src->info;
    dst->page_off = src->page_off;

    if (!src->info.is_leaf)
      memcpy(dst->childs, src->childs, (src->info.nentries + 1) *
             sizeof(offset_t));

    memcpy(dst->keys, src->keys, src->info.nentries * dbh->fheader.key_size);
    memcpy(dst->data_off, src->data_off,
           src->info.nentries * sizeof(offset_t));
    memcpy(dst->data_size, src->data_size,
           src->info.nentries * sizeof(offset_t));
}

//-----------------------------------------------------------------------
uint32 BinarySearch (db_page      *p,
                     uint32       page_size,
                     void         *key,
                     int32        key_size)
{
   int32 i = page_size/2,
         cmp,
         ini = 0,
         fin = page_size-1;

   if (!page_size)
       return -1;

   do {

      cmp = memcmp(key, (uint8 *)p->keys + i*key_size, key_size);

      if (!cmp)
        return i;
      else if (cmp > 0)
        ini = i+1;
      else
        fin = i-1;

      i = (ini + fin)/2;

   } while (ini <= fin);

   return -(ini+1);
}
//-----------------------------------------------------------------------
offset_t SeekEntry(db_handle    *dbh,
                   void         *key,
                   void *data)
{
   db_page *p = AllocPage(dbh);
   int i;

   if (!p)
      return 0;

   LoadPage(dbh, p, dbh->fheader.root_off);

   for(;;)
   {
       i = BinarySearch (p, p->info.nentries, key,
                         dbh->fheader.key_size);

       if (i >= 0)
       {
           fseek(dbh->fdata,p->data_off[i], SEEK_SET);
           fread(data, 1, dbh->fheader.data_size ?
                 dbh->fheader.data_size : p->data_size[i],
                 dbh->fdata);
           FreePage(p);
           return 1;
       }

       if (p->info.is_leaf)
       {
           FreePage(p);
           return 0;
       }


       i = -i-1;

       LoadPage(dbh, p, p->childs[i]);
   }
}
//-----------------------------------------------------------------------
void DataBufferInsert(db_handle *dbh, offset_t data_off, void *data, int data_size)
{
    if (dbh->data_cache.nentries == MAX_DATA_CACHE)
       DataBufferFlush(dbh);

    SL_Insert(&dbh->data_cache, &data_off, data, data_size, NULL);
}
//-----------------------------------------------------------------------
void DataBufferFlush(db_handle *dbh)
{
   sl_entry *pe, *pkill;

   pe = dbh->data_cache.levels->next[0];

   while (pe)
   {
      pkill = pe;
      pe = pe->next[0];

      fseek(dbh->fdata, *(offset_t *)pkill->key, SEEK_SET);
      fwrite(pkill->data, dbh->fheader.data_size ?
             dbh->fheader.data_size : pkill->data_size, 1,
             dbh->fdata);
      fflush(dbh->fdata);
      SL_FreeEntry(pkill);
   }

   dbh->data_cache.max_level = 0;
   dbh->data_cache.nentries = 0;
   dbh->data_cache.levels->next[0] = NULL;
}
//-----------------------------------------------------------------------
offset_t InsertEntry(db_handle  *dbh,
                     void       *key,
                     void       *data,
                     uint32     data_size,
                     void       (*pfun)(void *, int, void *, int)
                     )
{
   db_page *p = AllocPage(dbh);
   db_page *psplit;
   int i, j;
   offset_t pointer, offset, data_offset;
   stack_t stack;
   void *key_cpy = malloc(dbh->fheader.key_size);

   if (!p || !key_cpy)
   {
      if (p)
        FreePage(p);

      return 0;
   }

   memcpy(key_cpy, key, dbh->fheader.key_size);

   InitStack(&stack, sizeof(offset_t));

   LoadPage(dbh, p, dbh->fheader.root_off);

   for(;;)
   {
       i = BinarySearch (p, p->info.nentries, key_cpy,
                        dbh->fheader.key_size);

       if (i >= 0)
       {
           if (pfun)
           {
             void *read_data;
             int sz;

             if ((sz = SL_Find(&dbh->data_cache, p->data_off+i, &read_data)))
                pfun(data, dbh->fheader.data_size ? 
                     dbh->fheader.data_size : data_size, read_data, sz);
             else
             {
                read_data = malloc(dbh->fheader.data_size ? 
                            dbh->fheader.data_size : p->data_size[i]);

                if (read_data)
                {
                   fseek(dbh->fdata,p->data_off[i], SEEK_SET);
                   fread(read_data, 1, dbh->fheader.data_size ? 
                       dbh->fheader.data_size : p->data_size[i], dbh->fdata);

                   pfun(data, dbh->fheader.data_size ? 
                       dbh->fheader.data_size : data_size, read_data, 
                       dbh->fheader.data_size ? dbh->fheader.data_size :
                       p->data_size[i]);

                   DataBufferInsert(dbh, p->data_off[i], read_data, 
                                    dbh->fheader.data_size ?
                                    dbh->fheader.data_size : 
                                    data_size);

                   free(read_data);
                }
             }
           }
           FreePage(p);
           free(key_cpy);
           ClearStack(&stack);
           return 0; // already stored
       }

       i = -i-1;

       if (p->info.is_leaf)
           break;

       InsertStack(&stack, &(p->page_off));
       InsertStack(&stack, &(i));

       LoadPage(dbh, p, p->childs[i]);
   }

   data_offset = dbh->data_off;
   dbh->data_off += dbh->fheader.data_size ? dbh->fheader.data_size :
                    data_size;

   DataBufferInsert(dbh, data_offset, data, dbh->fheader.data_size ?
       dbh->fheader.data_size : data_size);

   pointer = 0;

   for(;;) {

       for (j = p->info.nentries-1; j >= i; j--)
       {
           memcpy((uint8*)p->keys +(j+1)*dbh->fheader.key_size,
                  (uint8*)p->keys +j*dbh->fheader.key_size,
                  dbh->fheader.key_size);

           p->childs[j+2] = p->childs[j+1];
           p->data_off[j+1] = p->data_off[j];
           p->data_size[j+1] = p->data_size[j];
       }

       memcpy((uint8*)p->keys + i*dbh->fheader.key_size,
              key_cpy, dbh->fheader.key_size);


       p->childs[i+1] = pointer;
       p->data_off[i] = data_offset;
       p->data_size[i] = data_size;

       if (p->info.nentries < dbh->fheader.page_size)
       {
          p->info.nentries++;
          StorePage(dbh,p);
          FreePage(p);
          ClearStack(&stack);
          free(key_cpy);
          return 1;  // new_entry
       }

       // size == max_size, we need to fragment this page...

       i = p->info.nentries/2;

       psplit = AllocPage(dbh);

       FillPage(dbh, psplit, dbh->fheader.key_size,
                ((uint8*)p->keys+(i+1)*dbh->fheader.key_size),
                p->data_off+i+1, p->data_size+i+1, i,
                p->childs+i+1);

       psplit->info.is_leaf = p->info.is_leaf;
       p->info.nentries = i;

       for(j = i+1; (unsigned)j <= dbh->fheader.page_size; j++)
             p->childs[j] = 0;

       memcpy(key_cpy, (uint8*)p->keys+i*dbh->fheader.key_size,
              dbh->fheader.key_size);

       data_offset = p->data_off[i];
       data_size = p->data_size[i];

       if (p->page_off == dbh->fheader.root_off) // raiz
       {
          db_page *new_root = AllocPage(dbh);

          new_root->info.is_leaf = 0;
          new_root->info.nentries = 0;
          new_root->page_off = 0;
          new_root->childs[0] = p->page_off;
          StorePage(dbh,new_root);
          dbh->fheader.root_off = new_root->page_off;
          InsertStack(&stack, &(new_root->page_off));
          InsertStack(&stack, &(new_root->info.nentries));
          FreePage(new_root);
       }

       StorePage(dbh,psplit);
       StorePage(dbh,p);

       pointer = psplit->page_off;
       ExtractStack(&stack, &i);
       ExtractStack(&stack, &offset);
       LoadPage(dbh, p, offset);
       FreePage(psplit);
   } 
}
//------------------------------------------------------
offset_t DeleteEntry(db_handle  *dbh,
                     void       *key)
{
   db_page *p = AllocPage(dbh);
   db_page *neighbour, *parent;
   int i, j;
   unsigned anchor;
   offset_t parent_off;
   uint32 limit = (dbh->fheader.page_size+1)/2;
   stack_t stack;


   if (!p)
      return 0;

   InitStack(&stack, sizeof(offset_t));
   LoadPage(dbh, p, dbh->fheader.root_off);

   for(;;)
   {
       i = BinarySearch (p, p->info.nentries, key,
                         dbh->fheader.key_size);

       if (i >= 0)
          break;

       if (p->info.is_leaf)
       {
           FreePage(p);
           ClearStack(&stack);
           return -1;
       }

       i = -i-1;

       InsertStack(&stack, &(p->page_off));
       InsertStack(&stack, &i);

       LoadPage(dbh, p, p->childs[i]);
   }

   if (!p->info.is_leaf)
   {
       // we go down to the leaves to find the biggest key
       // from the smaller keys subtree.

       parent = p;
       p = AllocPage(dbh);

       InsertStack(&stack, &(parent->page_off));
       InsertStack(&stack, &i);
       LoadPage(dbh, p, parent->childs[i]);

       while (!p->info.is_leaf)
       {
          InsertStack(&stack, &(p->page_off));
          InsertStack(&stack, &(p->info.nentries));
          LoadPage(dbh, p, p->childs[p->info.nentries]);
       }

       memcpy((uint8*)parent->keys+i*dbh->fheader.key_size,
              (uint8*)p->keys + (p->info.nentries-1)*dbh->fheader.key_size,
              dbh->fheader.key_size);

       parent->data_off[i] = p->data_off[p->info.nentries-1];
       parent->data_size[i] = p->data_size[p->info.nentries-1];

       StorePage(dbh,parent);
       FreePage(parent);

       i = p->info.nentries-1;
   }


   for(j=i; (unsigned)j < p->info.nentries; j++)
   {
      memcpy((uint8*)p->keys+j*dbh->fheader.key_size,
             (uint8*)p->keys+(j+1)*dbh->fheader.key_size,
             dbh->fheader.key_size);

      p->data_off[j] = p->data_off[j+1];
      p->data_size[j] = p->data_size[j+1];
   }

   p->info.nentries--;

   for(;;)
   {
     // if no underflow ocurs or we are at the root node
     // we've finished.

     if (p->info.nentries >= limit ||
         p->page_off == dbh->fheader.root_off)
     {
         if (p->page_off == dbh->fheader.root_off &&
             p->info.nentries == 0 && !p->info.is_leaf)
         {
             dbh->fheader.root_off = p->childs[0];
         }
         else
             StorePage(dbh,p);
         FreePage(p);
         ClearStack(&stack);
         return 0;
     }

     // now we have to see if we can take one entry from
     // one of the neighbour nodes without leaving it
     // below the min-limit

     neighbour = AllocPage(dbh);
     parent = AllocPage(dbh);

     ExtractStack(&stack, &anchor);
     ExtractStack(&stack, &parent_off);

     LoadPage(dbh, parent, parent_off);

     // neighbour to the left...

     if (anchor != 0)
     {
         LoadPage(dbh, neighbour, parent->childs[anchor-1]);
         if (neighbour->info.nentries > limit)
         {
             for (j = p->info.nentries-1; j >= 0; j--)
             {
                memcpy((uint8*)p->keys +(j+1)*dbh->fheader.key_size,
                       (uint8*)p->keys +j*dbh->fheader.key_size,
                       dbh->fheader.key_size);

                p->childs[j+2] = p->childs[j+1];
                p->data_off[j+1] = p->data_off[j];
                p->data_size[j+1] = p->data_size[j];
             }

             p->childs[1] = p->childs[0];
             p->childs[0] = neighbour->childs[neighbour->info.nentries];

             memcpy(p->keys,
                    (uint8*)parent->keys + (anchor-1) * dbh->fheader.key_size,
                    dbh->fheader.key_size);

             p->data_off[0] = parent->data_off[anchor-1];
             p->data_size[0] = parent->data_size[anchor-1];

             p->info.nentries++;

             memcpy((uint8*)parent->keys + (anchor-1) * dbh->fheader.key_size,
                    (uint8*)neighbour->keys + (neighbour->info.nentries-1) *
                    dbh->fheader.key_size, dbh->fheader.key_size);

             parent->data_off[anchor-1] = neighbour->data_off[neighbour->info.nentries-1];
             parent->data_size[anchor-1] = neighbour->data_size[neighbour->info.nentries-1];

             neighbour->info.nentries--;

             StorePage(dbh, p);
             StorePage(dbh, parent);
             StorePage(dbh, neighbour);
             FreePage(p);
             FreePage(parent);
             FreePage(neighbour);
             return 0;
         }
     }

     // neighbour to the right...

     if (anchor != parent->info.nentries)
     {
         LoadPage(dbh, neighbour, parent->childs[anchor+1]);
         if (neighbour->info.nentries > limit)
         {
             memcpy((uint8*)p->keys + p->info.nentries * dbh->fheader.key_size,
                    (uint8*)parent->keys + anchor * dbh->fheader.key_size,
                    dbh->fheader.key_size);

             p->data_off[p->info.nentries] = parent->data_off[anchor];
             p->data_size[p->info.nentries] = parent->data_size[anchor];

             p->info.nentries++;
             p->childs[p->info.nentries] = neighbour->childs[0];

             memcpy((uint8*)parent->keys + anchor * dbh->fheader.key_size,
                    (uint8*)neighbour->keys, dbh->fheader.key_size);

             parent->data_off[anchor] = neighbour->data_off[0];
             parent->data_size[anchor] = neighbour->data_size[0];

             for(j=0; (unsigned)j < neighbour->info.nentries; j++)
             {
                memcpy((uint8*)neighbour->keys+j*dbh->fheader.key_size,
                       (uint8*)neighbour->keys+(j+1)*dbh->fheader.key_size,
                       dbh->fheader.key_size);
                neighbour->childs[j] = neighbour->childs[j+1];
                neighbour->data_off[j] = neighbour->data_off[j+1];
                neighbour->data_size[j] = neighbour->data_size[j+1];
             }

             neighbour->info.nentries--;
             StorePage(dbh, p);
             StorePage(dbh, parent);
             StorePage(dbh, neighbour);
             FreePage(p);
             FreePage(parent);
             FreePage(neighbour);
             return 0;
         }
     }

     // Last alternative: merge current page with one of its
     // neighbours and the anchor entry from the parent node

     if (anchor == parent->info.nentries)
     {
         swap(p, neighbour, db_page *);
         anchor--;
     }

     memcpy((uint8*)p->keys + p->info.nentries * dbh->fheader.key_size,
            (uint8*)parent->keys + anchor * dbh->fheader.key_size,
            dbh->fheader.key_size);

     p->data_off[p->info.nentries] = parent->data_off[anchor];
     p->data_size[p->info.nentries] = parent->data_size[anchor];

     for (j = p->info.nentries+1, i = 0;(unsigned) i < neighbour->info.nentries; 
          i++, j++)
     {
          memcpy((uint8*)p->keys + j * dbh->fheader.key_size,
                 (uint8*)neighbour->keys + i * dbh->fheader.key_size,
                 dbh->fheader.key_size);
          p->childs[j] = neighbour->childs[i];
          p->data_off[j] = neighbour->data_off[i];
          p->data_size[j] = neighbour->data_size[i];
     }
     p->childs[j] = neighbour->childs[i];

     for (j = anchor; (unsigned)j < parent->info.nentries; j++)
     {
          memcpy((uint8*)parent->keys + j * dbh->fheader.key_size,
                 (uint8*)parent->keys + (j+1) * dbh->fheader.key_size,
                 dbh->fheader.key_size);
          parent->childs[j+1] = parent->childs[j+2];
          parent->data_off[j] = parent->data_off[j+1];
          parent->data_size[j] = parent->data_size[j+1];
     }

     p->info.nentries += neighbour->info.nentries + 1;
     parent->info.nentries--;
     StorePage(dbh,p);
     FreePage(neighbour);
     FreePage(p);
     p = parent;
   }
}
//-----------------------------------------------------------------------
void StorePage(db_handle  *dbh,
               db_page    *p)
{
    unsigned int pipe, key;

    if (!p->page_off)
    {
        p->page_off = dbh->fheader.file_size;
        dbh->fheader.file_size += PAGE_SIZE(dbh);
    }

    key = hash_fun(p->page_off);

    for(pipe = 0; pipe < dbh->npipes; pipe++)
    {
       if (dbh->cache[pipe][key].page->page_off == p->page_off)
       {
          ClonePage(dbh, dbh->cache[pipe][key].page, p);
          dbh->cache[pipe][key].dirty = 1;
          return;
       }
    }

    pipe = dbh->npipes-1;

    if (dbh->cache[pipe][key].dirty)
    {
       WritePageToDisk(dbh,dbh->cache[pipe][key].page);
       dbh->cache[pipe][key].dirty = 0;
    }

    while (pipe > 0)
    {
       swap(dbh->cache[pipe], dbh->cache[pipe-1], cache_entry *);
       pipe--;
    }

    ClonePage(dbh, dbh->cache[0][key].page,p);
    dbh->cache[0][key].dirty = 1;
}
//------------------------------------------------------
void WritePageToDisk(db_handle  *dbh,
                     db_page    *p)
{
    fseek(dbh->fp, p->page_off, SEEK_SET);
    fwrite(&p->info, sizeof(page_header), 1, dbh->fp);
    fwrite(p->keys, dbh->fheader.key_size, dbh->fheader.page_size, dbh->fp);
    fwrite(p->childs, sizeof(offset_t), dbh->fheader.page_size+1, dbh->fp);
    fwrite(p->data_off, sizeof(offset_t), dbh->fheader.page_size, dbh->fp);
    if (!dbh->fheader.data_size)
       fwrite(p->data_size, sizeof(offset_t), dbh->fheader.page_size, dbh->fp);
    fflush(dbh->fp);
}
//------------------------------------------------------
int32 num_pages, num_entries;

void PrintTreeStats(db_handle *dbh)
{
   db_page *p = AllocPage(dbh);

   num_entries = 0;
   num_pages   = 0;

   if (p)
   {
      LoadPage(dbh, p, dbh->fheader.root_off);
      PrintInPreOrder(dbh,p);
      FreePage(p);

      printf("\n\t*** DB stats ***\n");
      printf("\tnum_entries: %8d\n",num_entries);
      printf("\tnum_pages:   %8d\n",num_pages);
      printf("\tfill ratio:  %8.2f %%\n\n",(num_entries*100.0/num_pages)/dbh->fheader.page_size);
   }
}
//------------------------------------------------------
void PrintInPreOrder(db_handle *dbh, db_page *p)
{
   unsigned int i;

   num_pages++;
   num_entries += p->info.nentries;

   if (!p->info.is_leaf)
   {
       db_page *child = AllocPage(dbh);

       for(i=0; i <= p->info.nentries; i++)
       {
          LoadPage(dbh, child, p->childs[i]);
          PrintInPreOrder(dbh,child);
       }

       FreePage(child);
   }
}
//------------------------------------------------------
void InitCache(db_handle *dbh, int cache_size, int npipes)
{
    int i, j;

    dbh->cache_pipe_size = cache_size;
    dbh->npipes = npipes;

    dbh->cache = malloc(npipes * sizeof(cache_entry *));

    for(j=0; j < npipes; j++)
    {
       dbh->cache[j] = malloc(cache_size*sizeof(cache_entry));
       for (i=0; (unsigned)i < dbh->cache_pipe_size; i++)
       {
           dbh->cache[j][i].page = AllocPage(dbh);
           dbh->cache[j][i].page->page_off = 0;
           dbh->cache[j][i].dirty = 0;
       }
    }
}
//------------------------------------------------------
void CloseCache(db_handle *dbh)
{
    unsigned int i, j;

    for(j=0; j < dbh->npipes; j++)
    {
      for (i=0; i < dbh->cache_pipe_size; i++)
      {
          if (dbh->cache[j][i].dirty)
              WritePageToDisk(dbh,dbh->cache[j][i].page);

          FreePage(dbh->cache[j][i].page);
      }
      free(dbh->cache[j]);
    }

    free(dbh->cache);
}
//--------------------------------------
file_run *AllocNewRun(bulk_load_info *info, int nrun)
{
   file_run *fr = malloc(sizeof(file_run));

   if (fr)
   {
       fr->fname = malloc(20);

       if (!fr->fname)
       {
           free(fr);
           return NULL;
       }

       if (nrun)
         sprintf(fr->fname, "file%d.tmp", nrun);
       else
         sprintf(fr->fname, "file%d.tmp", info->nruns);

       info->nruns++;

       fr->fp = fopen(fr->fname, "wb");

       if (!fr->fp)
       {
           free(fr->fname);
           free(fr);
           return NULL;
       }

       fr->buffer = NULL;
       fr->next = NULL;
   }
   return fr;        
}
//--------------------------------------
void FreeRun(file_run *fr)
{
   if (fr)
   {
       if (fr->fp)
           fclose(fr->fp);

       if (fr->fname)
       {
          remove(fr->fname);
          free(fr->fname);
       }

       if (fr->buffer)
           free(fr->buffer);

       free(fr);
   }
}
//--------------------------------------
int InitBulkLoad(bulk_load_info *info, int key_size)
{
    SL_Init(&info->buffer, key_size);

    info->nruns = 0;
    info->key_size = key_size;
    info->last_written = NULL;
    info->current_run = AllocNewRun(info, 0);
    return info->current_run ? 0 : -1;
}
//--------------------------------------
void CloseCurrentRun(bulk_load_info *info, int alloc_new)
{
   sl_entry *e;

   if (info->last_written)
   {
       SL_FreeEntry(info->last_written);
       info->last_written = NULL;
   }

   while ((e = SL_ExtractFirst(&info->buffer)))
   {
      fwrite(e->key, info->key_size, 1, info->current_run->fp);
      fwrite(e->data, e->data_size, 1, info->current_run->fp);
      SL_FreeEntry(e);
   }

   fclose(info->current_run->fp);

   if (alloc_new)
   {
      file_run *fr = info->current_run;
      info->current_run = AllocNewRun(info, 0);
      info->current_run->next = fr;
   }
}
//--------------------------------------
int MergeTempFiles(bulk_load_info *info, void (*pfun)(void *, int, void *, int),
                   int (*confirm_pfun)(void *data, int data_size))
{
    file_run *curr, *aux, *min, *fr_ini;
    int cmp;
    int m, lim;
    int merge, last_step = 0;
    int rejected = 0;
    FILE *fdest;


    CloseCurrentRun(info, 0);

    printf("Merging temporary files...\n");

    SL_DestroyList(&info->buffer);

    fr_ini = NULL;
    merge = info->nruns;

    do
    {
       lim = MERGE_WAYS;

       if (!fr_ini)
       {
           fr_ini = info->current_run;
           info->current_run = NULL;

           if (info->nruns <= MERGE_WAYS)
           {
               last_step = 1;
               printf("\nlast step...");
           }
       }

       aux = info->current_run;
       info->current_run = AllocNewRun(info, merge++);
       info->current_run->next = aux;
       fdest = info->current_run->fp;

       printf("\n%10s <- ", info->current_run->fname);


       for (curr = fr_ini, m = 0, aux = NULL; curr && m < lim; aux = curr, curr = curr->next, m++)
       {
          curr->buffer = malloc(INPUT_BUFFER_SIZE * (info->data_size + info->key_size));
          curr->fp = fopen(curr->fname,"rb");
          curr->nbuffer = 0;
          curr->prev = aux;
          if (m)
             printf(" + %10s",curr->fname);
          else
             printf(" %10s",curr->fname);
       }

       do {
             curr = fr_ini;
             m = 0;
             min = NULL;

             do
             {
                 if (!curr->nbuffer)
                 {
                    if (!curr->fp)
                    {
                        if (curr->prev)
                            curr->prev->next = curr->next;

                        if (curr->next)
                            curr->next->prev = curr->prev;

                        aux = curr;
                        curr = curr->next;

                        lim--;

                        if (aux == fr_ini)
                            fr_ini = curr;

                        FreeRun(aux);
                        info->nruns--;

                        continue;
                    }
                    else
                    {
                       curr->nbuffer = fread(curr->buffer, info->data_size + info->key_size, INPUT_BUFFER_SIZE, curr->fp);
                       curr->pbuffer = curr->buffer;

                       if (curr->nbuffer != INPUT_BUFFER_SIZE)
                       {
                           fclose(curr->fp);
                           curr->fp = NULL;

                           if (!curr->nbuffer)
                           {
                              if (curr->prev)
                                 curr->prev->next = curr->next;

                              if (curr->next)
                                 curr->next->prev = curr->prev;

                             aux = curr;
                             curr = curr->next;

                             lim--;

                             if (aux == fr_ini)
                               fr_ini = curr;

                             FreeRun(aux);
                             info->nruns--;
                             continue;
                           }
                       }
                    }
                 }

                 if (!min)
                     min = curr;
                 else
                 {
                     cmp = memcmp(curr->pbuffer, min->pbuffer, info->key_size);

                     if (cmp < 0)
                        min = curr;
                     else if (!cmp)
                     {
                        pfun(curr->pbuffer + info->key_size, info->data_size, 
                             min->pbuffer + info->key_size, info->data_size);

                        curr->pbuffer += info->data_size + info->key_size;
                        curr->nbuffer--;
                     }
                 }

                 curr = curr->next;
                 m++;

             } while (curr && m < lim);

             if (min)
             {
                if (!last_step || !confirm_pfun ||
                    confirm_pfun(min->pbuffer + info->key_size, info->data_size)
                   )
                {
                    fwrite(min->pbuffer, info->key_size + info->data_size, 1, fdest);
                }
                else
                    rejected++;

                min->pbuffer += info->data_size + info->key_size;
                min->nbuffer--;
             }

       } while (min);

       fclose(fdest);
    } while (info->nruns != 1);


//    Superb MVC6 bug; this line compiles :-)
//
//    printf("\n\nMerge process finished. Records sorted: %d. Rejected: %d\n", FileSize(info->current_run->fname)
//           / (info->data_size + info->key_size, rejected));

    printf("\n\nMerge process finished. Records sorted: %d. Rejected: %d\n", FileSize(info->current_run->fname)
           / (info->data_size + info->key_size), rejected);

    return 0;
}
//--------------------------------------------------------
offset_t GenerateRun(bulk_load_info *info,
                     void       *key,
                     void       *data,
                     uint32     data_size,
                     void       (*pfun)(void *, int, void *, int)
                     )
{
    SL_Insert(&info->buffer, key, data, data_size, pfun);

    if (info->buffer.nentries > MAX_RUN_BUFFER)
    {
        sl_entry *first = SL_ExtractFirst(&info->buffer);

        if(!memcmp(first->key, key, info->key_size) &&
            info->last_written && memcmp(key, info->last_written->key, info->key_size) <= 0)
        {
            CloseCurrentRun(info, 1);
            SL_Insert(&info->buffer, first->key, first->data, first->data_size, NULL);
            SL_FreeEntry(first);
        }
        else
        {
            fwrite(first->key, info->key_size, 1, info->current_run->fp);
            fwrite(first->data, first->data_size, 1, info->current_run->fp);
            if (info->last_written)
              SL_FreeEntry(info->last_written);
            info->last_written = first;
        }
    }

    return 0;
}

/*

  - Raiz:

  Tenemos N elementos ordenados. El objetivo es crear un arbol B de orden k
  (k punteros a paginas en cada pagina) con esos N elementos.

  Sea h la altura necesaria del arbol B, que es Techo(log k (N+1)). El numero
  de nodos en la raiz es nr = N - k ^ (h-1). Ahora lanzamos recursivamente una
  función para crear nr+1 árboles B de altura h-1 con la salvedad de que su raiz
  debe tener un minimo de Techo((k-1)/2)  nodos para lo cual hay que particionar
  en nr+1 grupos el conjunto inicial. Sea s = Suelo((N-nr)/(nr+1)) y r el resto de
  esa division. Entonces tendremos r grupos de s+1 elementos y  nr+1-r de s
  elementos. Puffff a ver como demoños organizo esto

*/

int BulkLoadTree(db_handle *dbh, char *fname)
{
   db_page *p = AllocPage(dbh);
   unsigned int s, r, lim, i, nr, max, min, height;
   offset_t data_offset;
   uint8 *buffer, *pbuffer;
   int nbuffer = 0;
   unsigned int num_ele = FileSize(fname)/(dbh->bload.data_size + dbh->bload.key_size);
   int unread = num_ele;
   FILE *fp;

   printf("Bulk-loading B-tree from %s...\n", fname);


   buffer = malloc(INPUT_BUFFER_SIZE * (dbh->bload.data_size + dbh->bload.key_size));
   fp = fopen(fname,"rb");

   dbh->data_off = 4;
   fseek(dbh->fdata,4,SEEK_SET);

   if (!p || !buffer || !fp)
   {
       if (p)
           FreePage(p);

       if (buffer)
           free(buffer);

       if (fp)
           fclose(fp);

       perror("Error");
       return -1;
   }

   if (num_ele <= dbh->fheader.page_size) // height = 1
   {
      p->info.is_leaf = 1;
      p->info.nentries = num_ele;

      for (i=0; i < num_ele; i++)
      {
          if (!nbuffer)
          {
             nbuffer = fread(buffer, dbh->bload.data_size + dbh->bload.key_size, Min(unread, INPUT_BUFFER_SIZE), fp);
             pbuffer = buffer;
             unread -= nbuffer;
          }

          data_offset = dbh->data_off;
          dbh->data_off += dbh->bload.data_size;

          memcpy((uint8*)p->keys + i * dbh->bload.key_size, pbuffer, dbh->bload.key_size);
          p->data_off[i] = data_offset;
          p->data_size[i] = dbh->bload.data_size;
          fwrite(pbuffer+dbh->bload.key_size, dbh->bload.data_size, 1, dbh->fdata);
          pbuffer += dbh->bload.key_size + dbh->bload.data_size;
      }

      p->page_off = 0;
   }
   else
   {
      p->info.is_leaf = 0;
      height = (int)ceil(logn(dbh->fheader.page_size+1, num_ele+1));

      max = (int)pow(dbh->fheader.page_size+1,height-1)-1;
      min = (int)pow((dbh->fheader.page_size+1)/2, height-1)-1;

      for (nr = 1; nr <= dbh->fheader.page_size; nr++)
      {
          s = (num_ele - nr)/(nr+1);
          r = (num_ele - nr)%(nr+1);

          if (s >= min && (s < max || (s == max && !r)))
              break;
      }

      assert(nr <= dbh->fheader.page_size);

      // we will have nr entries at this page, with nr+1 subtrees, r holding
      // s+1 entries and nr+1-r with s entries.

      p->info.nentries = nr;

      lim = Min(nr, r);

      for(i = 0; i < lim; i++)
      {
          p->childs[i] = GenerateSubTreeNonRoot(dbh, fp, s+1, height-1);

          fread((uint8*)p->keys + i*dbh->bload.key_size, dbh->bload.key_size, 1, fp);
          fread(buffer, dbh->bload.data_size, 1, fp);
          p->data_size[i] = dbh->bload.data_size;
          p->data_off[i] = dbh->data_off;
          dbh->data_off += dbh->bload.data_size;
          fwrite(buffer, dbh->bload.data_size, 1, dbh->fdata);
      }

      for(; i < nr ; i++)
      {
          p->childs[i] = GenerateSubTreeNonRoot(dbh, fp, s, height-1);

          fread((uint8*)p->keys + i*dbh->bload.key_size, dbh->bload.key_size, 1, fp);
          fread(buffer, dbh->bload.data_size, 1, fp);
          p->data_size[i] = dbh->bload.data_size;
          p->data_off[i] = dbh->data_off;
          dbh->data_off += dbh->bload.data_size;
          fwrite(buffer, dbh->bload.data_size, 1, dbh->fdata);
      }

      p->childs[i] = GenerateSubTreeNonRoot(dbh, fp, s + (lim == nr), height-1);
   }

   p->page_off = 0;
   StorePage(dbh,p);
   dbh->fheader.root_off = p->page_off;
   FreePage(p);
   fflush(dbh->fdata);
   fclose(fp);
   return 0;
}

//--------------------------------------------------------
offset_t GenerateSubTreeNonRoot(db_handle *dbh, FILE *fp, unsigned int num_ele, unsigned int height)
{
   db_page *p = AllocPage(dbh);
   unsigned int s, r, i, nr, lim, max, min;
   offset_t page_offset;
   offset_t data_offset;
   uint8 *buffer, *pbuffer;
   int nbuffer = 0;
   int unread = num_ele;

   buffer = malloc(INPUT_BUFFER_SIZE * (dbh->bload.data_size + dbh->bload.key_size));

   if (!p || !buffer)
   {
       if (p)
           FreePage(p);

       if (buffer)
           free(buffer);

       perror("Error in malloc");
       return 0;
   }

   if (height == 1)
   {
      assert(num_ele <= dbh->fheader.page_size);
      p->info.is_leaf = 1;
      p->info.nentries = num_ele;

      for (i=0; i < num_ele; i++)
      {
          if (!nbuffer)
          {
             nbuffer = fread(buffer, dbh->bload.data_size + dbh->bload.key_size, Min(unread, INPUT_BUFFER_SIZE), fp);
             pbuffer = buffer;
             unread -= nbuffer;
          }

          data_offset = dbh->data_off;
          dbh->data_off += dbh->bload.data_size;

          memcpy((uint8*)p->keys + i * dbh->bload.key_size, pbuffer, dbh->bload.key_size);
          p->data_off[i] = data_offset;
          p->data_size[i] = dbh->bload.data_size;
          fwrite(pbuffer+dbh->bload.key_size, dbh->bload.data_size, 1, dbh->fdata);
          pbuffer += dbh->bload.key_size + dbh->bload.data_size;
      }

      p->page_off = 0;
   }
   else
   {
      p->info.is_leaf = 0;

      max = (int)pow(dbh->fheader.page_size+1,height-1)-1;
      min = (int)pow((dbh->fheader.page_size+1)/2, height-1)-1;

      for (nr = (dbh->fheader.page_size+1)/2; nr <= dbh->fheader.page_size; nr++)
      {
          s = (num_ele - nr)/(nr+1);
          r = (num_ele - nr)%(nr+1);

          if (s >= min && (s < max || (s == max && !r)))
              break;
      }

      assert(nr <= dbh->fheader.page_size);

      // we will have nr entries at this page, with nr+1 subtrees, r holding
      // s+1 entries and nr+1-r with s entries.

      p->info.nentries = nr;

      lim = Min(nr, r);

      for(i = 0; i < lim; i++)
      {
          p->childs[i] = GenerateSubTreeNonRoot(dbh, fp, s+1, height-1);

          fread((uint8*)p->keys + i*dbh->bload.key_size, dbh->bload.key_size, 1, fp);
          fread(buffer, dbh->bload.data_size, 1, fp);
          p->data_size[i] = dbh->bload.data_size;
          p->data_off[i] = dbh->data_off;
          dbh->data_off += dbh->bload.data_size;
          fwrite(buffer, dbh->bload.data_size, 1, dbh->fdata);
      }

      for(; i < nr ; i++)
      {
          p->childs[i] = GenerateSubTreeNonRoot(dbh, fp, s, height-1);

          fread((uint8*)p->keys + i*dbh->bload.key_size, dbh->bload.key_size, 1, fp);
          fread(buffer, dbh->bload.data_size, 1, fp);
          p->data_size[i] = dbh->bload.data_size;
          p->data_off[i] = dbh->data_off;
          dbh->data_off += dbh->bload.data_size;
          fwrite(buffer, dbh->bload.data_size, 1, dbh->fdata);
      }

      p->childs[i] = GenerateSubTreeNonRoot(dbh, fp, s + (lim == nr), height-1);
   }

   p->page_off = 0;
   StorePage(dbh,p);
   page_offset = p->page_off;
   FreePage(p);
   free(buffer);
   return page_offset;
}
