/afs/cs.wisc.edu/u/n/w/nwilliam/private/workspace/Quut/src/index.cpp

#include <sstream>
#include <cstring>
#include <math.h>
#include "index.h"
#include "filescan.h"
#include "exceptions/bad_param_exception.h"
#include "exceptions/duplicate_attribute_exception.h"
#include "exceptions/no_more_records_exception.h"
#include "exceptions/end_of_file_exception.h"
#include "exceptions/file_not_found_exception.h"
#include "exceptions/record_not_found_exception.h"
#include "exceptions/duplicate_insertion_exception.h"
#include "exceptions/directory_overflow_exception.h"

#define MAX(a,b) ((a) > (b) ? (a) : (b))

namespace badgerdb {

// Constructor for the Index class. The arguments passed in are 
//     'name' -- file name of the relation
//     'offset', 'length', 'type' -- describing the attribute indexed
//     'unique' -- flag for enforcing uniquess of the entries

Index::Index(const std::string &name, int offset, int length, Datatype type, int unique, BufMgr *bufferMgr)
{
  bufMgr = bufferMgr;
  Page *pagePtr;

  // For this assignment turn off hash indices on strings
  if(type == STRING) 
  {
    throw BadParamException();
  }
   
  curBuc = NULL;

  recSize = length + sizeof(RecordId);     // size of the index entry
  numSlots = (Page::SIZE - 2*sizeof(short)) / recSize;   // # entries on a page

  // get name of the index file by concatenating relation name and
  // the offset of the attribute

  std::ostringstream outputString;
  outputString << name << '.' << offset << std::endl;
  std::string indexName = outputString.str(); 

  // The constructor runs in two cases. In the first case, an index 
  // doesn't exist as detected by db.openFile. The relation file 
  // should be scanned and each tuple is inserted to the index. In
  // the second case, an index already exists. Only the headerPage
  // needs to read in and pinned down in the buffer pool.

  try
  {
    file = new BlobFile(indexName, false/*open*/);  //open file

    headerPageNo = 1;
    bufMgr->readPage(file, headerPageNo, pagePtr);
    headerPage = (iHeaderPage*) pagePtr;
    dirSize = (int)pow(2, headerPage->depth);
    return;
  }
  catch(FileNotFoundException e)
  {
  }

  file = new BlobFile(indexName, true/*create*/); //create file

  // allocate and initialize the header page
  bufMgr->allocPage(file, headerPageNo, (Page*&)headerPage);
  strcpy(headerPage->fileName, name.c_str());

  headerPage->offset = offset;
  headerPage->length = length;
  headerPage->type = type;
  headerPage->depth = 0;
  headerPage->unique = unique;
  dirSize = 1;

  // allocate and initialize the first bucket pointed by dir[0]
  PageId pageNo;
  Bucket* bucket;
  bufMgr->allocPage(file, pageNo, (Page*&)bucket);

  bucket->depth = 0;
  bucket->slotCnt = 0;
  headerPage->dir[0] = pageNo;

  bufMgr->unPinPage(file, pageNo, true);

  // build index by scanning the relation file and inserting every
  // tuple

  FileScan fileScan(name, bufferMgr);

  fileScan.startScan(offset, length, type, std::string(""), EQ);

  RecordId   rid;

  while(1)
  {
    try
    {
      fileScan.scanNext(rid);
      std::string recordString = fileScan.getRecord();
      insertEntry(recordString.c_str() + offset, rid);
    }
    catch(EndOfFileException &e)
    {
      break;
    }
  }
}

Index::~Index() {

  bufMgr->unPinPage(file, headerPageNo, true);

  if (curBuc != NULL)
  {
    bufMgr->unPinPage(file, curPageNo, true);
  }
  
  bufMgr->flushFile(file);

  delete file;
}

// return a hashvalue hashed from the attribute 'value' according to 
// the current depth of the directory

void Index::hashIndex(const void *attr, int &hashvalue)
{
  int index;
  const char *value = (const char *)attr;

  switch (headerPage->type)
  {
    case STRING:
      {
        // For strings "add" the values of the characters, stopping 
        // at the first character that is 0
        index = value[0];
        for (int i = 1; i < headerPage->length; i++)
        {
          if (!value[i]) 
            index += value[i];
          else
            break;
        }
      }
      break;

    case DOUBLE:
      {
        // Treat doubles as strings of length sizeof (double)
        index = value[0];
        for (int i = 1; i < headerPage->length; i++) index += value[i];
      }
      break;

    case INTEGER:
      // copy the integer value to avoid byte-alignment errors.
      memcpy (&index, value, sizeof(int));
      break;
  }

  // hashvalue is the lower bits of 'index'. Number of bits is determined
  // by the depth of the directory

  index = std::abs(index);   
  int mask = 1 << headerPage->depth;
  hashvalue = index % mask;
}

// Insert an <attribute, rid> pair into the index. Return OK if the 
// entry is inserted and DIROVERFLOW if the directory isn't large
// enough to hold the indices.

void Index::insertEntry(const void *value, RecordId rid) 
{ 
  Bucket *bucket;
  Bucket *newBucket;
  PageId newPageNo;       
  char  data[Page::SIZE*2];
  int counter;
  int index;

  // If the 'unique' flag is set, scan the index to see if the
  // <attribute, rid> pair already exists

  if (headerPage->unique == UNIQUE)
  {
    RecordId outRid;

    startScan(value);

    while(1)
    {
      try
      {
        scanNext(outRid);
        if(!memcmp(&outRid, &rid, sizeof(RecordId)))
          throw DuplicateInsertionException();
      }
      catch (NoMoreRecordsException &e)
      {
        break;
      }
    }
    
    endScan();
  }
 
  // Get the bucket containing the entry into buffer pool
  hashIndex(value, index);
  PageId pageNo = headerPage->dir[index];

#ifdef DEBUGIND
  std::cout << "Inserting entry " << *(int*)value << " to bucket " 
    << pageNo << std::endl;
#endif //DEBUGIND
 
  bufMgr->readPage(file, pageNo, (Page*&)bucket);

  // the bucket needs to be splitted if the number of entries
  // on the bucket equals the maximum
  if (bucket->slotCnt == numSlots) { // splitting bucket
    
    // allocate a new bucket
    bufMgr->allocPage(file, newPageNo, (Page*&)newBucket);  

    // Initialize this newly allocated bucket
    newBucket->depth = ++(bucket->depth);
    newBucket->slotCnt = 0;

    // Copy all (value, rid) pairs in the old bucket and the new
    // entry to a temporary area
    memcpy(data, bucket->data, numSlots*recSize);
    memcpy(&(data[numSlots*recSize]), value, headerPage->length);
    memcpy(&(data[numSlots*recSize + headerPage->length]), &rid, sizeof(RecordId));

    counter = bucket->slotCnt + 1;
    bucket->slotCnt = 0;

    // the directory needs to be doubled if the depth of the bucket
    // being splitted equals the depth of the directory

    if (bucket->depth > headerPage->depth)
    { // doubling directory

      // The directory is doubled and the lower half of the directory
      // is copied to the upper half

      int newDirSize = 2 * dirSize;
      if (newDirSize > DIRSIZE)
        throw DirectoryOverflowException();

      for(int i = 0; i < dirSize; i++)
      {
        headerPage->dir[i + dirSize] = headerPage->dir[i];
      }

      dirSize = newDirSize;
      (headerPage->depth)++;
      headerPage->dir[index + (1 << (bucket->depth - 1))] = newPageNo;
    }
    else
    {
      // reset the appropriate directories to the new bucket
      int oldindex = index % (1 << (bucket->depth - 1));
      int newindex = oldindex + (1 << (bucket->depth - 1));
      for (int j = 0; j < dirSize; j++)
      {
        if ((j % (1 << (bucket->depth))) == newindex)
        {
          headerPage->dir[j] = newPageNo;
        }
      }
    }

#ifdef DEBUGIND
    printDir();
#endif //DEBUGIND

    // call insertEntry recursively to insert all (value, rid) 
    // pairs in the temporary area to the index

    for (int k = 0; k < counter; k++) {
      value = &(data[k * recSize]);
      rid = * ((RecordId *)((char *)value + headerPage->length));
      insertEntry(value, rid);
    }

    bufMgr->unPinPage(file, newPageNo, true);

  }
  else
  {
    // There is sufficient free space in the bucket. Insert (value, rid) here

    int offset = (bucket->slotCnt) * recSize;
    memcpy(&(bucket->data[offset]), value, headerPage->length);
    memcpy(&(bucket->data[offset+headerPage->length]), &rid, sizeof(RecordId));
    (bucket->slotCnt)++;
  }

  bufMgr->unPinPage(file, pageNo, true);
}

void Index::printDir() {
  std::cout << "printing directory...\n";
  std::cout << "depth is " << headerPage->depth << std::endl;
  for (int i = 0; i < dirSize; i++) 
    std::cout << i << "\tpoints to bucket " << headerPage->dir[i] << std::endl;
  std::cout << std::endl;
}

// delete <attribute, rid> pair from the index. return OK if deleted
// return RECNOTFOUND if such a pair does not exist in the index

void Index::deleteEntry(const void* value, const RecordId &rid)
{
  int index;
  Bucket* bucket;

  hashIndex(value, index);

  // read in the bucket that might have the entry in it
  PageId pageNo = headerPage->dir[index];
  bufMgr->readPage(file, pageNo, (Page*&)bucket);

  // scan the bucket for the entry. Delete it if found
  for(int i = 0; i < bucket->slotCnt; i++)
  {
    try
    {
      matchRec(bucket, value, i);
      if(!memcmp(&rid, &(bucket->data[i*recSize + headerPage->length]), sizeof(RecordId)))
      {
        // the entry is found. Decrease the entry counts in the bucket
        // and copy the last entry in the bucket to the slot occupied
        // by the deleted entry

        (bucket->slotCnt)--;
        memcpy(&(bucket->data[i*recSize]), &(bucket->data[recSize*(bucket->slotCnt)]), recSize);
        bufMgr->unPinPage(file, pageNo, true);
        return;
      }
    }
    catch(RecordNotFoundException &e)
    {
    }
  }

  bufMgr->unPinPage(file, pageNo, false);

  throw RecordNotFoundException();
}

// Test if the index entry in 'bucket' matches the attribute 'value.
// 'offset' indicates the location of the entry in the bucket

void Index::matchRec(const Bucket* bucket, const void* value, int offset) 
{
  const char* tmp = &(bucket->data[offset*recSize]);

  switch(headerPage->type)
  {
  case INTEGER:
    if (*(int *)value == *(int *)tmp)
    //if (atoi((char *)value) == *(int *)tmp)
      return;
    break;

  case DOUBLE:
    {
       // Take care of byte misallignments 
       double inVal, inTmp;
       memcpy (&inVal, value, sizeof (double));
       memcpy (&inTmp, tmp, sizeof (double));
       // if (*(double *)value == *(double*)tmp) return OK;
       if (inVal == inTmp) return;
       //if (atof((char *)value) == inTmp) return;
    }
    break;

  case STRING:
    if (!memcmp(value, tmp, headerPage->length))
      return;
    break;
  }

  throw RecordNotFoundException();
}

void Index::printBucs()
{

  Bucket* bucket;

  std::cerr << std::endl << "printing indices"<< std::endl;

  for (int i = 0; i < dirSize; i++)
  {
    PageId pageNo = headerPage->dir[i];
    std::cerr << "page " << pageNo << ": ";

    bufMgr->readPage(file, pageNo, (Page*&)bucket);
    for (int j = 0; j < bucket->slotCnt; j++)
    {
      switch(headerPage->type)
      {
        case INTEGER:
          std::cerr << *(int*)&(bucket->data[j*recSize]) << "\t";
          break;
        case DOUBLE:
          std::cerr << *(double*)&(bucket->data[j*recSize]) << "\t";
          break;
        default:break;
      }
    }
    std::cerr << std::endl;
    bufMgr->unPinPage(file, pageNo, false);
  }
}

// start a scan of the entries with attribute 'value'. return SCANTABFULL
// if too many scans are open at the same time

void Index::startScan(const void* scanValue)
{
  int hashvalue;

  curValue = scanValue;

  // read in and pin down the bucket containing entries with 'value'
  // in the buffer pool
  hashIndex(curValue, hashvalue);

  PageId pageNo = curPageNo = headerPage->dir[hashvalue];
  bufMgr->readPage(file, pageNo, (Page*&)curBuc);

  curOffset = 0;
}

// return the next entry with attribute 'value'. return NOMORERECS if
// it reaches the end of the bucket

void Index::scanNext(RecordId& outRid) 
{
  int &offset = curOffset;
  Bucket* buc = curBuc;
  const void* value = curValue;

  while(offset < buc->slotCnt)
  {
    try
    {
      matchRec(buc, value, offset);
      break;
    }
    catch(RecordNotFoundException &e)
    {
    }

    offset++;
  }

  if (offset == buc->slotCnt)
    throw NoMoreRecordsException();
  else
  {
    outRid = *(RecordId *)&(buc->data[offset*recSize + headerPage->length]);
    offset++;
  }
}

// unpin the bucket and clear the scan table entry.
void Index::endScan() 
{
  if (curBuc != NULL)
  {
    bufMgr->unPinPage(file, curPageNo, false);
    curBuc = NULL;
  }
}

/*
// find a free entry in the scan table
int Index::findFree()
{
  int i;
  for(i = 0; i < MAXOPENSCANS; i++)
  {
    if (openScans[i].valid == false) return i;
  }
  return -1;
}
*/

}