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

#include <iostream>
#include <cstdlib>
#include <sstream>
#include <cstring>
#include <vector>

#include "exceptions/end_of_file_exception.h"
#include "exceptions/bad_param_exception.h"
#include "exceptions/file_exists_exception.h"

#include "sort.h"
#include "buffer.h"

#define MIN(a,b)   ((a) < (b) ? (a) : (b))


// These comparison functions are visible only within this
// source file. reccmp is the comparison routine (much like
// strcmp or memcmp) that accepts integers, floats, and strings.
// It returns -1 if p1 is less than p2, +1 if p1 is greater
// than p2, or zero otherwise.

extern badgerdb::BufMgr *bufMgr;  // pointer to the attribute catalogs

namespace badgerdb
{

static int reccmp(char* p1, char* p2, int p1Len, int p2Len, Datatype type)
{
  float diff = 0.0;

  switch(type) {
  case INTEGER:
    int iattr, ifltr;                   // word-alignment problem possible
    memcpy(&iattr, p1, sizeof(int));
    memcpy(&ifltr, p2, sizeof(int));
    diff = iattr - ifltr;
    break;
  case DOUBLE:
    double fattr, ffltr;                 // word-alignment problem possible
    memcpy(&fattr, p1, sizeof(double));
    memcpy(&ffltr, p2, sizeof(double));
    diff = fattr - ffltr;
    break;
  case STRING:
    diff = memcmp(p1, p2, MIN(p1Len, p2Len));
    break;
  default:
    break;
  }

  if(diff < 0)
    diff = -1;
  else if (diff > 0)
    diff = 1;

  return (int)diff;
}


// These three comparison routines are jacketed versions of
// reccmp. This is because qsort(3) takes only a function pointer
// but no additional parameters. The objects pointed to by p1
// and p2 are of type SORTREC which has a pointer to the field
// to be compared as well as its length (used for strings).

#define SR(p)  ((SORTREC*)p)

static int intcmp(const void* p1, const void* p2)
{
 p1 = (char*) p1;
 p2 = (char*) p2;
  return reccmp(SR(p1)->field, SR(p2)->field, SR(p1)->length, SR(p2)->length, INTEGER);
}

static int floatcmp(const void* p1, const void* p2)
{
 p1 = (char*) p1;
 p2 = (char*) p2;
  return reccmp(SR(p1)->field, SR(p2)->field, SR(p1)->length, SR(p2)->length, DOUBLE);
}


static int stringcmp(const void* p1, const void* p2)
{
 p1 = (char*) p1;
 p2 = (char*) p2;
  return reccmp(SR(p1)->field, SR(p2)->field, SR(p1)->length, SR(p2)->length, STRING);
}


// Create a sorted temporary file of the source file (fileName).
// Sorting is based on attribute that is defined by offset, len,
// and type. maxItems is the maximum number of items that a sorted
// sub-run can hold (usually derived from amount of memory available).
// Status code is returned in variable status.

SortedFile::SortedFile(const std::string &fileName, int offset, int len, Datatype type, int maxItems)
 : fileName(fileName), type(type), offset(offset), length(len), maxItems(maxItems)
{
  if(offset < 0 || len < 1)
    throw BadParamException();

  if(type != STRING && type != INTEGER && type != DOUBLE)
    throw BadParamException();

  if((type == INTEGER && len != sizeof(int)) || (type == DOUBLE && len != sizeof(double)))
    throw BadParamException();

  // Must have space for at least 2 items (records) because otherwise
  // items cannot be swapped and sorted!

  if(maxItems < 2)
    throw BadParamException();

  buffer = new SORTREC [maxItems];
    
  sortFile();
}


// Sort file into sub-runs. The source file is split into runs
// which have at most maxItems records each. That many records
// are read into memory, sorted using qsort(3), and then written
// to a temporary file.

void SortedFile::sortFile()
{
  std::string rec;

  // Open source file.

  // Start an unfiltered sequential scan.

  file = new FileScan(fileName, bufMgr);
  file->startScan(0, 0, STRING, std::string(""), EQ);

  // As long as the source file has more records, collect up to
  // maxItems records into buffer and then dump records into
  // temporary file.

  do
  {
    for(numItems = 0; numItems < maxItems; numItems++)
    {
      // Fetch next record from source file, check if end of file.

      try
      {
        file->scanNext(buffer[numItems].rid);
      }
      catch(EndOfFileException &e)
      {
        break;
      }

      rec = file->getRecord();

      // Create space for holding a copy of the sorting attribute
      // only (rest of record is read when temporary file is
      // written). Copy sorting attribute from source record and
      // store the length of the attribute (reccmp is general-
      // purpose and can be shared by multiple instances of
      // SortedFile!).

      buffer[numItems].field = new char [length];

      memcpy(buffer[numItems].field, (char *)rec.c_str() + offset, length);
      buffer[numItems].length = length;
    }

    // If at least 1 record in sub-run, sort records and write out
    // to temporary file.

    if (numItems > 0)
    {
      generateRun(numItems);

      for(int i = 0; i < numItems; i++)
        delete [] buffer[i].field;
    }
  }while(numItems > 0);

  // Terminate sequential scan on source file and close file.

  delete file;

  // Prepare a sequential scan on each sub-run so that next()
  // can fetch next record from each run.

  startScans();
}


// Sort the records in buffer[] (actually, the sorting attribute
// plus the associated RecordId) and then dump records into temporary
// file.

void SortedFile::generateRun(int items)
{
  // Sort buffer using library function qsort (quick sort). Use
  // the appropriate comparison function for integers, floats,
  // or strings (qsort can't take type as a parameter).

  if (type == INTEGER)
    qsort(buffer, items, sizeof(SORTREC), intcmp);
  else if (type == DOUBLE)
    qsort(buffer, items, sizeof(SORTREC), floatcmp);
  else
    qsort(buffer, items, sizeof(SORTREC), stringcmp);

  // If this is the first sub-run, malloc space for a RUN object,
  // otherwise realloc more space. Note that on most systems
  // realloc(NULL) could be used even when runs == NULL, but
  // this doesn't work on all systems.

  RUN newRun;
  runs.push_back(newRun);

  RUN &run = runs.back();

  //Generate file name for temporary file.
  std::ostringstream outputString;
  outputString << fileName << ".sort." << runs.size();
  run.name = outputString.str();

#ifdef DEBUGSORT
  std::cout << "%%  Writing " << items << " tuples to file " << run->name
       << std::endl;
#endif

  // Make sure temporary file does not exist already. We don't
  // want to corrupt somebody else's sorted files (on another
  // attribute, for example).

  if(File::exists(run.name))
    throw FileExistsException(run.name);

  // Open a heap file. This will also create the temporary file.

  run.file = new FileScan(run.name, bufMgr);
  //run.file->startScan(0, 0, STRING, std::string(""), EQ);

  // Open an unfiltered sequential scan on the source file. The scan
  // is not actually needed for anything else than just getting
  // a scanId which getRandomRecord requires.

  FileScan scan(fileName, bufMgr);
  //scan.startScan(0, 0, STRING, std::string(""), EQ);

  // For each sort record (attribute plus RecordId) in the buffer, fetch
  // the whole record from the source file and then insert it into
  // the temporary file.

  for(int i = 0; i < items; i++)
  {
    SORTREC* rec = &buffer[i];
    RecordId rid;
    std::string record;

    record = scan.getRandomRecord(rec->rid);

    run.file->insertRecord(record, rid);
  }

  delete run.file;
}


// Prepare a sequential scan on each sub-run so that next()
// can fetch the next record from each run. The valid bit of
// each run is marked false to indicate that the (first)
// record has not been fetched yet. next() must therefore
// fetch it.

void SortedFile::startScans()
{
  std::vector<RUN>::iterator run;

#ifdef DEBUGSORT
  std::cout << "Start scan on : ";
#endif
  for(run = runs.begin(); run != runs.end(); run++)
  {
    run->file = new FileScan(run->name, bufMgr);
    //run->file->startScan(0, 0, STRING, std::string(""), EQ);
#ifdef DEBUGSORT
    std::cout << run->name << ", " << std::endl;
#endif

    run->valid = false;
    run->rid.page_number = 0;
    run->rid.slot_number = 0;
    run->mark.page_number = 0;
    run->mark.slot_number = 0;
  }
#ifdef DEBUGSORT
  std::cout << std::endl;
#endif
}


// Retrieve the next smallest record from the set of sorted sub-runs.
// The next record of each sub-run is peeked to find out the
// smallest of all. The pointer in the chosen sub-run is then
// advanced.

std::string SortedFile::next()
{
  // Empty source file has zero sub-runs and causes
  // end of file to be returned.

  if(runs.size() <= 0)
    throw EndOfFileException();

  // Find the run which has the smallest next record. If a run
  // has false valid bit, it doesn't have the next record in memory
  // yet.

  int smallest = -1;
  std::vector<RUN>::iterator run;

  for(run = runs.begin(); run != runs.end(); run++)
  {
    if (run->valid == false)
    {          // no record fetched yet for this run?
      try
      {
        run->file->scanNext(run->rid);
        run->rec = run->file->getRecord();
      }
      catch(EndOfFileException &e)
      {
        run->rid.page_number = 0;           // mark end of file
      }
      run->valid = true;                // a record is now in memory
    }

    if(run->rid.page_number == 0)            // end of run already?
      continue;

    if (smallest == -1)                     // select first one as smallest
      smallest = run - runs.begin();
    else if (reccmp((char *)runs[smallest].rec.c_str() + offset, (char *)run->rec.c_str() + offset, length, length, type) > 0)
      smallest = run- runs.begin();                   // current run has smaller next
  }

  if (smallest == -1)                        // no next record found?
    throw EndOfFileException();

#ifdef DEBUGSORT
  std::cout << "%%  Retrieved smallest from " << runs[smallest].name << std::endl;
#endif

  runs[smallest].valid = false;              // must fetch new record next time

  return runs[smallest].rec;               // give record pointers to caller
}


// Remember a position in the sorted output so that the caller
// can later return to this spot. The current RecordId of each sub-run
// is recorded in a separate field.

void SortedFile::setMark()
{
#ifdef DEBUGSORT
  std::cout << "%%  Setting mark in file" << std::endl;
#endif

  std::vector<RUN>::iterator run;

  for(run = runs.begin(); run != runs.end(); run++)
  {
    run->mark.page_number = run->rid.page_number;
    run->mark.slot_number = run->rid.slot_number;
#ifdef DEBUGSORT
    std::cout << "%%  Run " << i << " is at page " << run->mark.page_number << ", slot " << run->mark.slot_number << std::endl;
#endif
  }
}


// Restore sort position by fetching the last marked record
// of each sub-run using getRandomRecord. This allows the caller
// to back up in the sorted sequence (used in sort-merge join
// in case of duplicates).

void SortedFile::gotoMark()
{
#ifdef DEBUGSORT
  std::cout << "%%  Going to a mark in file" << std::endl;
#endif

  std::vector<RUN>::iterator run;

  for(run = runs.begin(); run != runs.end(); run++)
  {
    run->rid.page_number = run->mark.page_number;
    run->rid.slot_number = run->mark.slot_number;

#ifdef DEBUGSORT
    if(run->rid.page_number > 0)
      std::cout << "%%  Run " << i << " going to page " << run->rid.page_number << ", slot " << run->rid.slot_number << std::endl;
    else
      std::cout << "%%  Run " << i << " is already at eof" << std::endl;
#endif

    // Restore file position only if last marked position is
    // something else than end of file.

    if(run->rid.page_number > 0)
    {
      run->rec = run->file->getRandomRecord(run->rid);
      run->file->gotoMark(run->rid);
    }

    // Current record is already in memory so next() must not
    // advance in the temporary file.

    run->valid = true;
  }
}


// Deallocate all space allocated for this sorted file and
// delete temporary files.

SortedFile::~SortedFile()
{
  for(unsigned int i = 0; i < runs.size(); i++)
  {
    delete runs[i].file;
    File::remove(runs[i].name);
  }

  delete [] buffer;
}

}