SimpleDB.h

Go to the documentation of this file.

// -*- C++ -*-

// SimpleDB.cc: Simple Database Representation (implementation)
//
// Copyright (C) 2000 Nicolas Chapados
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
// 
//  1. Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
// 
//  2. Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in the
//     documentation and/or other materials provided with the distribution.
// 
//  3. The name of the authors may not be used to endorse or promote
//     products derived from this software without specific prior written
//     permission.
// 
// THIS SOFTWARE IS PROVIDED BY THE AUTHORS ``AS IS'' AND ANY EXPRESS OR
// IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
// OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
// NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
// TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
// 
// This file is part of the PLearn library. For more information on the PLearn
// library, go to the PLearn Web site at www.plearn.org

#ifndef SIMPLEDB_H
#define SIMPLEDB_H


#include <limits.h>                          
#include <string>
#include <vector>
#include <iostream>
//#include <iomanip>
#include <stdexcept>         
#include <typeinfo>
//#include <algorithm>

// norman: added check for WIN32
#ifndef WIN32
#include <unistd.h>                          
#else
#include <assert.h>
#include <io.h>
#endif

#include <fcntl.h>                           
#include <errno.h>                           
#include <stdlib.h>                          
//#include <ctype.h>


//#include "general.h"
#include <plearn/base/stringutils.h>     
//#include "random.h"
#include <plearn/base/PDate.h>                       
#include <plearn/math/Hash.h>
#include <plearn/base/TinyVector.h>

#ifdef WIN32
// norman: potentially dangerous if there is a function called with the same name in this
//         file. Beware!
#define open _open
#define close _close
#define lseek _lseek
#define read _read
#define write _write
#define unlink _unlink
#endif

namespace PLearn {
using namespace std;

class Row;
    

//##############################  CLASS  FIELD  ###########################
//              StringType              length of string
//                                      (including null terminator)
//              CharacterType           1 == unsigned char
//              SignedCharType          1 == signed char
//              ShortType               2 == signed short
//              IntType                 4 == signed int
//              FloatType               4 == float
//              DoubleType              8 == double
//              DateType                4 == PDate
enum FieldType {
  Unknown = 0,
  StringType,
  CharacterType,
  SignedCharType,
  ShortType,
  IntType,
  FloatType,
  DoubleType,
  DateType
};

struct Field {
  Field() : name(), field_type(Unknown), precision() {}
  Field(string name, FieldType t, int p = 0)
    : name(name), field_type(t)
    {
      switch(field_type) {
      case Unknown:             precision=0;    break;
      case StringType:  precision=p;    break;
      case CharacterType:       precision=1;    break;
      case SignedCharType:      precision=1;    break;
      case ShortType:           precision=2;    break;
      case IntType:             precision=4;    break;
      case FloatType:           precision=4;    break;
      case DoubleType:  precision=8;    break;
      case DateType:            precision=4;    break;
      default:
        PLERROR("Unknown field type %d with name %s",
                int(field_type), name.c_str());
      }
      if (sizeof(PDate) != 4)
        PLERROR("A PLearn PDate must have sizeof equal to 4");
    }

  bool operator==(const Field& x) const {
    return name==x.name &&
      field_type == x.field_type &&
      precision == x.precision;
  }
        
  string name;                       
  FieldType field_type;
  int precision;
};

extern const char MissingString;
extern const unsigned char MissingCharacter;
extern const signed char MissingSignedChar;
extern const short MissingShort;
extern const int MissingInt;
extern const float MissingFloat;
extern const double MissingDouble;
extern const PDate MissingDate;
    

//#############################  CLASS  FIELDPTR  #########################
class FieldPtr {
  friend class Row;                          
  friend class Schema;               

public:
  FieldPtr() : field_index_(-1), offset_(-1) {}


  int field_index() const {
    return field_index_;
  }

  ptrdiff_t offset() const {
    return offset_;
  }
        
  operator bool() const {
    return field_index_ >= 0;
  }

  bool operator!() const {
    return field_index_ == -1;
  }

  bool operator==(const FieldPtr& x) const {
    return field_index_ == x.field_index_ &&
      offset_ == x.offset_;
  }

  bool operator!=(const FieldPtr& x) const {
    return !(*this == x);
  }

private:
  FieldPtr(int fi, ptrdiff_t o) : field_index_(fi), offset_(o) {}

private:
  int field_index_;                  
  ptrdiff_t offset_;                 
};


//############################  CLASS  FIELDVALUE  ##########################
class FieldValue
{
  friend class FieldRowRef;
    
public:
  FieldValue();
  FieldValue(const FieldValue& fv);
  ~FieldValue();

  explicit FieldValue(const char*);
  explicit FieldValue(unsigned char);
  explicit FieldValue(signed char);
  explicit FieldValue(short);
  explicit FieldValue(int);
  explicit FieldValue(float);
  explicit FieldValue(double);
  explicit FieldValue(const PDate&);

  bool isMissing() const;
  void setMissing();
    
  string toString() const;
  double toDouble() const;
  PDate  toDate() const;

  operator double() const { return toDouble(); }
  operator float() const { return float(toDouble()); }
  operator int() const { return int(toDouble()); }
  operator string() const { return toString(); }
  operator PDate() const { return toDate(); }
    

  FieldValue& operator=(FieldValue);

  bool operator==(const FieldValue&) const;
  bool operator< (const FieldValue&) const;

  bool operator!=(const FieldValue& rhs) const {
    return !(*this == rhs);
  }
    
  bool operator<=(const FieldValue& rhs) const {
    return (*this == rhs) || (*this < rhs);
  }
    
  bool operator> (const FieldValue& rhs) const {
    return !(*this <= rhs);
  }
    
  bool operator>=(const FieldValue& rhs) const {
    return !(*this < rhs);
  }

  FieldValue operator+(const FieldValue&) const;
  FieldValue operator-(const FieldValue&) const;
  FieldValue operator*(const FieldValue&) const;
  FieldValue operator/(const FieldValue&) const;
    
  void swap(FieldValue&);

  bool isString() const {
    return field_type_ == StringType;
  }
    
  bool isIntegral() const {
    return
      field_type_ == CharacterType || field_type_ == SignedCharType ||
      field_type_ == ShortType     || field_type_ == IntType;
  }
    
  bool isFloating() const {
    return field_type_ == FloatType || field_type_ == DoubleType;
  }
    
  bool isDate() const {
    return field_type_ == DateType;
  }
    
private:
  FieldType field_type_;
  int precision_;
  struct DateVal_t {
    short year;
    unsigned char month, day;
  };
  union {
    char* string_val_;
    long long_val_;
    double double_val_;
    struct DateVal_t date_val_;
  };
};

ostream& operator<<(ostream&, const FieldValue&);



//##############################  CLASS  SCHEMA  ############################
class Schema : public vector<Field>
{
  typedef vector<Field> inherited;
  typedef Field T;

 public:
  Schema() : inherited() {}
  Schema(size_type n, const T& value) : inherited(n, value) {}
  Schema(int n, const T& value) : inherited(n, value) {}
  Schema(long n, const T& value) : inherited(n, value) {}
  explicit Schema(size_type n) : inherited(n) {}
  Schema(const Schema& x) : inherited(x) {}

#ifdef __STL_MEMBER_TEMPLATES
  template <class InputIterator>
    Schema(InputIterator first, InputIterator last)
    : inherited(first, last) {}
#else 
  Schema(const_iterator first, const_iterator last)
    : inherited(first, last) {}
#endif 

        


  bool findColumn(const string& name, int& position, int& start,
                  int& precision) const;

  FieldPtr findColumn(int position) const;

  FieldPtr findColumn(const string& name) const;

  FieldPtr operator()(int position) const {
    return findColumn(position);
  }

  FieldPtr operator()(const string& name) const {
    return findColumn(name);
  }
};


  
//#########################  CLASS  SIMPLEDBINDEXKEY  #######################
template <class KeyType>
class SimpleDBIndexKey {
public:
  typedef KeyType ByteArr;
  typedef typename ByteArr::iterator iterator;

  SimpleDBIndexKey() {}              
            
  explicit SimpleDBIndexKey(size_t len)
    : raw(len, '\0') {}
            
  SimpleDBIndexKey(const unsigned char* the_raw, size_t len)
    : raw(len, '\0')
    {
      copy(the_raw, the_raw+len, begin());
    }
                
  SimpleDBIndexKey(const ByteArr& the_raw)
    : raw(the_raw) {}

  operator char*() const {
    return (char*)(&raw[0]);
  }

  size_t byteLength() const {
    return raw.size();
  }

  bool operator==(const SimpleDBIndexKey& other) const {
    return raw == other.raw;
  }

  bool operator!=(const SimpleDBIndexKey& other) const {
    return raw != other.raw;
  }

  void resize(size_t len) {
    raw.resize(len);
  }

  typename ByteArr::iterator begin() {
    return raw.begin();
  }

  typename ByteArr::iterator end(){
    return raw.end();
  }

private:
  ByteArr raw;
};

  
  
//#############################  CLASS  SIMPLEDB  #########################
template <class KeyType = TinyVector<unsigned char, 8>,
          class QueryResult = TinyVector<unsigned int, 4> >
class SimpleDB
{

public:

  typedef unsigned long RowNumber;
  enum {
    InvalidRow = ULONG_MAX 
  };

  typedef unsigned long Offset;

  enum AccessType {
    readwrite = 0,
    readonly  = 1
  };
        
  static const Offset AbsoluteFileLimit = 512ul * 1024ul * 1024ul - 1;
        
  typedef QueryResult QueryResult_t;         
  static QueryResult EmptyResult;            
    
  typedef SimpleDBIndexKey<KeyType> IndexKey;
  typedef Hash<IndexKey,QueryResult> Index;
  typedef PP<Index> PIndex;
    
public:

  SimpleDB(string rootname, string path=".", AccessType = readwrite,
           bool verbose=true);
  virtual ~SimpleDB();

        
  string getName() const {
    return name;
  }
  string getPath() const {
    return path;
  }
        
        
  void setSchema(const Schema& s);

  const Schema& getSchema() const {
    return schema;
  }

  void saveSchema();
  void loadSchema();
      
  bool findColumn(string name, int& position, int& start,
                  int& precision) const {
    return schema.findColumn(name,position,start,precision);
  }

  int indexOfField(const string& fieldname) const {
    return schema(fieldname).field_index();
  }

  Row getRow(RowNumber) const;
  Row& getInRow(RowNumber, Row&) const;
  RowNumber size() const { return size_; }  

  int length() const {
    return int(size());
  }
  int width() const {
    return int(schema.size());
  }

  void addRow(const Row&);                   
  void setRow(const Row&, RowNumber);        
  void truncateFromRow(RowNumber n);         




  bool indexColumn(string columnName,
                   string secondColumn = string(""));

  void clearIndex(string columnName);
    
  QueryResult findEqual(const unsigned char* lookfor,
                        string columnName,
                        string secondColumn = string(""));

  const QueryResult& findEqualIndexed(const unsigned char* lookfor,
                                      string columnName,
                                      string secondColumn = string(""));

  QueryResult findEqualLinear(const unsigned char* lookfor,
                              string columnName,
                              string secondColumn = string(""));
        
  typedef vector<const unsigned char*> vuc;
  QueryResult findEqualLinear(const vuc& lookfor,
                              string columnName,
                              string secondColumn = string(""));

  double tableSizeMultiplier() const {
    return table_size_multiplier;
  }
  void tableSizeMultiplier(double x) {
    table_size_multiplier = x;
  }
    
        
private:
  void computeSize();

  void memoryToDisk(Row&) const;

  void diskToMemory(Row&) const;

  int seekToRow(RowNumber) const;

  int seekToEnd() const;

  void openAllFiles() const;
        
  void closeAllFiles() const;

  inline int lastSegment() const;
        
  string getSegmentPath(int i) const;
        
private:
  string name;                       
  string path;                       
  AccessType access_type;                    
  int access_mask;                           
  RowNumber size_;                           


  Schema schema;                             
  int row_size;                      
  RowNumber max_records_file;                

        
  mutable vector<int> allfd;                 



  double table_size_multiplier;
    
  vector<PIndex> indexes;

  bool verbose;                      

private:
  SimpleDB(const SimpleDB&);
  void operator=(const SimpleDB&);
};


//###########################  CLASS  ROWITERATOR  ##########################
//                      *myit1 = *myit2;

class FieldRowRef;
  
class RowIterator {
public:
  RowIterator() : curfield(0), curptr(0), schema(0) { }
  RowIterator(const RowIterator& x) {
    curfield = x.curfield;
    curptr = x.curptr;
    schema = x.schema;
  }
  bool operator==(const RowIterator& x) {
    return
      curfield==x.curfield &&
      curptr  ==x.curptr   &&
      schema  ==x.schema;
  }
  bool operator!=(const RowIterator& x) {
    return !((*this) == x);
  }

  RowIterator(int curf, unsigned char* curp, const Schema* sc)
    : curfield(curf), curptr(curp), schema(sc)
    { }


  RowIterator& operator=(const RowIterator& x) 
    {
      if (&x != this) 
      {
        curfield = x.curfield;
        curptr   = x.curptr;
        schema   = x.schema;
      }
      return *this;
    }
            
  void copyFrom(const RowIterator& it)
    {
#ifdef BOUNDCHECK
      if(it.precision()!=precision() || it.getFieldType()!=getFieldType())
        PLERROR("In Row::iterator::copyFrom Source and destination fields not of same type or precision");
#endif
      copy(it.raw(),it.raw()+it.precision(),raw());
    }

  inline FieldRowRef operator*() const;
    

  RowIterator& operator++() {
    if (schema && curptr && curfield < schema->size()) {
      curptr += (*schema)[curfield].precision;
      ++curfield;
    }
    return *this;
  }

  RowIterator operator++(int) {
    RowIterator x = *this;
    ++(*this);
    return x;
  }

  RowIterator operator[](int i) {
    assert(i >= 0);
    RowIterator it = *this;
    while (i--)
      ++it;
    return it;
  }

  FieldType getFieldType() const
    { return (*schema)[curfield].field_type; }


  bool isString() const {
    return schema && curptr && curfield < schema->size() &&
      (*schema)[curfield].field_type == StringType;
  }
            
  bool isCharacter() const {
    return schema && curptr && curfield < schema->size() &&
      (*schema)[curfield].field_type == CharacterType;
  }
            
  bool isSignedChar() const {
    return schema && curptr && curfield < schema->size() &&
      (*schema)[curfield].field_type == SignedCharType;
  }
            
  bool isShort() const {
    return schema && curptr && curfield < schema->size() &&
      (*schema)[curfield].field_type == ShortType;
  }
            
  bool isInt() const {
    return schema && curptr && curfield < schema->size() &&
      (*schema)[curfield].field_type == IntType;
  }
            
  bool isFloat() const {
    return schema && curptr && curfield < schema->size() &&
      (*schema)[curfield].field_type == FloatType;
  }
            
  bool isDouble() const {
    return schema && curptr && curfield < schema->size() &&
      (*schema)[curfield].field_type == DoubleType;
  }

  bool isDate() const {
    return schema && curptr && curfield < schema->size() &&
      (*schema)[curfield].field_type == DateType;
  }
            
            
  char* asString() {
    bool iss = isString();
    if (iss)
      return reinterpret_cast<char*>(curptr);
    else 
      return 0;
    //return isString()? reinterpret_cast<char*>(curptr) : 0;
  }

  unsigned char* asCharacter() {
    return isCharacter()?
      reinterpret_cast<unsigned char*>(curptr) : 0;
  }

  signed char* asSignedChar() {
    return isSignedChar()?
      reinterpret_cast<signed char*>(curptr) : 0;
  }

  short* asShort() {
    return isShort()? reinterpret_cast<short*>(curptr) : 0;
  }

  int* asInt() {
    return isInt()? reinterpret_cast<int*>(curptr) : 0;
  }

  float* asFloat() {
    return isFloat()? reinterpret_cast<float*>(curptr) : 0;
  }

  double* asDouble() {
    return isDouble()? reinterpret_cast<double*>(curptr) : 0;
  }

  PDate* asDate() {
    return isDate()? reinterpret_cast<PDate*>(curptr) : 0;
  }


  const char* asString() const {
    return isString()? reinterpret_cast<const char*>(curptr) : 0;
  }

  const unsigned char* asCharacter() const {
    return isCharacter()?
      reinterpret_cast<const unsigned char*>(curptr) : 0;
  }

  const signed char* asSignedChar() const {
    return isSignedChar()?
      reinterpret_cast<const signed char*>(curptr) : 0;
  }

  const short* asShort() const {
    return isShort()? reinterpret_cast<const short*>(curptr) : 0;
  }

  const int* asInt() const {
    return isInt()? reinterpret_cast<const int*>(curptr) : 0;
  }

  const float* asFloat() const {
    return isFloat()? reinterpret_cast<const float*>(curptr) : 0;
  }

  const double* asDouble() const {
    return isDouble()? reinterpret_cast<const double*>(curptr) : 0;
  }

  const PDate* asDate() const {
    return isDate()? reinterpret_cast<const PDate*>(curptr) : 0;
  }


  double toDouble() const;
  string toString() const;

            
  bool isMissing() const;
  void setMissing();
            

  string name() const {
    return (schema && curfield < schema->size())?
      (*schema)[curfield].name : string("");
  }
            
  int precision() const {
    return (schema && curfield < schema->size())?
      (*schema)[curfield].precision : -1;
  }

  int char_width() const;

  unsigned char* raw() {
    return curptr;
  }

  const unsigned char* raw() const {
    return curptr;
  }

  FieldPtr makePtr() const {
    return schema->findColumn(curfield);
  }

private:
  unsigned curfield;                 
  unsigned char* curptr;             
  const Schema* schema;
};


//###########################  CLASS  FIELDROWREF  ##########################
class FieldRowRef
{
public:
  FieldRowRef(const RowIterator& it)
    : it_(it) {}

  operator FieldValue() const;
  FieldRowRef& operator=(const FieldValue&);
  FieldRowRef& operator=(const FieldRowRef rhs) {
    return operator=(FieldValue(rhs));
  }

  inline RowIterator operator&() const;
    
private:
  RowIterator it_;                           

};
  

//###############################  CLASS  ROW  ############################
class Row {
        
public:
  typedef RowIterator iterator;
  typedef iterator const_iterator;           
  typedef long size_type;                    

public:
  Row() : rawrow(), schema(0) { }
  Row(const Row& r) : rawrow(r.rawrow), schema(r.schema) { }

  Row(const Schema* s);
        
  Row(const vector<unsigned char>& raw, const Schema* s)
    : rawrow(raw), schema(s) { }
        
        
        
        
  iterator begin() {
    return iterator(0, raw(), schema);
  }

  iterator end() {
    if (schema)
      return iterator(schema->size(), raw()+rawrow.size(),
                      schema);
    else
      return iterator(0,0,0);
  }

  size_type size() const {
    return (size_type)rawrow.size();
  }
        
  size_type max_size() const {
    return (size_type)rawrow.size();
  }

  bool empty() const {
    return (schema && schema->empty()) || !schema;
  }

  const unsigned char* raw() const {
    if (rawrow.size())
      return &rawrow[0];
    else
      return 0;
  }

  unsigned char* raw() {
    if (rawrow.size())
      return &rawrow[0];
    else
      return 0;
  }

  const Schema* getSchema() const {
    return schema;
  }

  iterator operator[](int fieldNumber);
  iterator operator[](string fieldName);

  iterator bind(const FieldPtr& p) const {
    if (!p)
      PLERROR("Trying to dereference a null FieldPtr");
    return iterator(p.field_index_,
                    const_cast<unsigned char*>(raw()) + p.offset_,
                    schema);
  }

  void sanitize() const;

private:
  vector<unsigned char> rawrow;
  const Schema* schema;
};


//#####  Row-related global functions  ####################################

double todouble(const Row::iterator& it);
string tostring(const Row::iterator& it);
    
ostream& operator<<(ostream& o, const Row::iterator& field);

ostream& operator<<(ostream&, const Row& row);

void printFieldName(ostream& o, const Row::iterator& field);

void printFieldNames(ostream& o, const Row& row);


//#####  Miscellaneous Declarations  ########################################
  
typedef SimpleDB<> SDB;

void randomShuffleRows(SDB& sdb);

void halfShuffleRows(SDB& sdb);



//#####  Non-Template Inline Functions  #####################################

FieldRowRef RowIterator::operator*() const
{
  return FieldRowRef(*this);
}

RowIterator FieldRowRef::operator&() const
{
  return it_;
}

  
  
//#####  Implementation of Templates  #######################################

template <class KT, class QR>    
QR SimpleDB<KT,QR>::EmptyResult;



//#####  SimpleDB-related functions  ########################################
    
template <class KT, class QR>   
SimpleDB<KT,QR>::SimpleDB(string rootname, string the_path,
                          AccessType the_access_type, bool the_verbose)
  : name(rootname), path(the_path), access_type(the_access_type),
    access_mask(0), schema(), row_size(), allfd(),
    table_size_multiplier(1.8), indexes(), verbose(the_verbose)
{
  if (path != "")
    path += slash;
  string fullpath = path + name + ".sdb";

  switch (access_type) {
  case readwrite:
    access_mask = O_RDWR;
    break;
  case readonly:
    access_mask = O_RDONLY;
    break;
  }
        
  loadSchema();
  openAllFiles();
  computeSize();
}

template <class KT, class QR>   
SimpleDB<KT,QR>::~SimpleDB()
{
  closeAllFiles();
  saveSchema();
}

template <class KT, class QR>   
void SimpleDB<KT,QR>::setSchema(const Schema& s)
{
  schema = s;
  Row row(&s);
  row_size = row.size();
  indexes.resize(s.size());

  if (row_size > 0)
    max_records_file = RowNumber(AbsoluteFileLimit / row_size);
  else
    max_records_file = 0;


  closeAllFiles();
  openAllFiles();
  computeSize();
}

template <class KT, class QR>   
void SimpleDB<KT,QR>::saveSchema()
{
  if (access_type==readwrite) {
    string fullpath = path + name + ".ssc";
    ofstream sf(fullpath.c_str(), ios::out);
    Schema::iterator it = schema.begin(), end = schema.end();
    for (; it != end; ++it) {
      sf << it->name << " ";
      switch (it->field_type) {
      case Unknown:
        break;
      case StringType:
        sf << "string " << it->precision << endl;
        break;

      case CharacterType:
        sf << "character" << endl;
        break;
                
      case SignedCharType:
        sf << "signedchar" << endl;
        break;
            
      case ShortType:
        sf << "short" << endl;
        break;
            
      case IntType:
        sf << "int" << endl;
        break;
            
      case FloatType:
        sf << "float" << endl;
        break;
            
      case DoubleType:
        sf << "double" << endl;
        break;
            
      case DateType:
        sf << "date" << endl;
        break;
            
      default:
        PLERROR("Unknown field type in database: %d", it->field_type);
      }
    }
  }
}

template <class KT, class QR>   
void SimpleDB<KT,QR>::loadSchema()
{
  string fullpath = path + name + ".ssc";
  ifstream sf(fullpath.c_str());
  Schema schema;
  while (sf) {
    string name,type;
    sf >> name >> type;
    if (name.size() == 0 || type.size() == 0)
      break;
    type = lowerstring(type);
    if (type == "string") {
      int length;
      sf >> length;
      schema.push_back(Field(name,StringType,length));
    }
    else if (type == "character")
      schema.push_back(Field(name,CharacterType));
    else if (type == "signedchar")
      schema.push_back(Field(name,SignedCharType));
    else if (type == "short")
      schema.push_back(Field(name,ShortType));
    else if (type == "int")
      schema.push_back(Field(name,IntType));
    else if (type == "float")
      schema.push_back(Field(name,FloatType));
    else if (type == "double")
      schema.push_back(Field(name,DoubleType));
    else if (type == "date")
      schema.push_back(Field(name,DateType));
    else {
      cerr << "Unexpected input type \"" << type
           << "\" in schema file " << fullpath << endl;
      exit(1);
    }
  }
  setSchema(schema);
}

template <class KT, class QR>   
void SimpleDB<KT,QR>::addRow(const Row& row)
{
  if(row_size != row.size())
    PLERROR("In addRow row_size != row.size() (%d != %d)", row_size, row.size());
  row.sanitize();
  int fd = seekToEnd();
  off_t curpos = lseek(fd, 0L, SEEK_CUR);
    
#ifdef LITTLEENDIAN

  //ssize_t writtensize = ::write(fd, row.raw(), row_size);
  int writtensize = ::write(fd, row.raw(), row_size);
#endif    
#ifdef BIGENDIAN
  Row newrow(row);
  memoryToDisk(newrow);
  int writtensize = ::write(fd, newrow.raw(), row_size);
#endif

  if (writtensize == -1) {
#if defined(_MINGW_) || defined(WIN32)
    PLWARNING("could not truncate database file, end may be corrupted!");
#else
    ftruncate(fd, curpos);
#endif
    PLERROR("Could not write to database: %s", strerror(errno));
  }
  else
    size_++; 
}

template <class KT, class QR>   
void SimpleDB<KT,QR>::setRow(const Row& row, RowNumber n)
{
  if(n<0 || n>=size())
    PLERROR("Out of bounds in SimpleDB::setRow");
  if(row_size != row.size())
    PLERROR("In setRow row_size != row.size() (%d != %d)", row_size, row.size());
  row.sanitize();
  int fd = seekToRow(n);
        
#ifdef LITTLEENDIAN
  int writtensize = ::write(fd, row.raw(), row_size);
#endif
#ifdef BIGENDIAN
  Row newrow(row);
  memoryToDisk(newrow);
  int writtensize = ::write(fd, newrow.raw(), row_size);
#endif

  if (writtensize == -1)
    PLERROR("Could not write to database: %s",strerror(errno));
}

  
template <class KT, class QR>   
void SimpleDB<KT,QR>::truncateFromRow(RowNumber n)
{
  int curfd = seekToRow(n);
  off_t curpos = lseek(curfd, 0L, SEEK_CUR);
  if (ftruncate(curfd, curpos) == -1) {
    PLERROR((string("Could not truncate database at row ") +
             tostring(n) + ": " + strerror(errno)).c_str());
  }

  vector<int>::iterator found = find(allfd.begin(), allfd.end(), curfd);
  int fromfd = found-allfd.begin() + 1;
  int last = lastSegment();

  closeAllFiles();
  bool allok = true;
    
  for ( ; fromfd <= last; ++fromfd) {
    string path = getSegmentPath(fromfd);
    if(unlink(path.c_str()) == -1) {
      PLWARNING((string("Could not unlink database segment ") + path +
                 ": " + strerror(errno)).c_str());
      allok = false;
    }
  }
    
  if (allok) {
    openAllFiles();
    computeSize();
  }
  else
    PLERROR("Error during truncation");
}

  
template <class KT, class QR>   
Row& SimpleDB<KT,QR>::getInRow(RowNumber n, Row& row) const
{
  if(n<0 || n>=size())
    PLERROR("Out of Bounds in SimpleDB::getInRow"); 
  if(row_size != row.size())
    PLERROR("In getInRow row_size!=row_size()");
  int fd = seekToRow(n);

  int size_read = ::read(fd, row.raw(), row_size);
  if (size_read == -1)
    PLERROR("Could not read from database: %s",strerror(errno));
  diskToMemory(row);
  return row;
}

template <class KT, class QR>   
Row SimpleDB<KT,QR>::getRow(RowNumber n) const
{
  Row row(&schema);
  getInRow(n, row);
  return row;
}

template <class KT, class QR>   
void SimpleDB<KT,QR>::computeSize()
{ 
  if(row_size<=0)
    size_ = 0;
  else
  {
    size_ = 0; 
    int i=0;
    int bytesinfile = file_size(getSegmentPath(i++));
    while(bytesinfile>0)
    {
      size_ += bytesinfile/row_size;
      bytesinfile = file_size(getSegmentPath(i++));
    }
  }


}

template <class KT, class QR>   
void SimpleDB<KT,QR>::memoryToDisk(Row& row) const
{
#ifdef LITTLEENDIAN

#endif    
#ifdef BIGENDIAN
  Row newr(row);
  Row::iterator it = newr.begin(), end = newr.end();
  for(; it != end; ++it) {
    if (short* x = it.asShort())
      reverse_short(x,1);
    if (int* x = it.asInt())
      reverse_int(x,1);
    if (float* x = it.asFloat())
      reverse_float(x,1);
    if (double* x = it.asDouble())
      reverse_double(x,1);
    if (PDate* x = it.asDate()) {
      reverse_short(&(x->year),1);
    }
  }
#endif
}

template <class KT, class QR>   
void SimpleDB<KT,QR>::diskToMemory(Row& row) const
{
  memoryToDisk(row);
}


//#####  Physical Splitting Among Multiple Files  #########################

template <class KT, class QR>
int SimpleDB<KT,QR>::seekToRow(RowNumber i) const
{
      
  if (max_records_file == 0)
    PLERROR("Attempting to seekToRow without schema set");
        
  int segmentNumber = int(i / max_records_file);
  Offset rowInSegment = Offset(i % max_records_file);

  if (segmentNumber > lastSegment()) {
    for (int i = lastSegment()+1; i <= segmentNumber; ++i) {
      int fd = open(getSegmentPath(i).c_str(),
                    access_mask | O_CREAT, 0777);
      if (fd == -1)
        PLERROR("Could not open database segment %d at path %s: %s",
                i, getSegmentPath(i).c_str(), strerror(errno));
      allfd.push_back(fd);
    }
  }
  if (allfd[segmentNumber] == -1)
    PLERROR("Problem accessing database segment %d at path %s",
            segmentNumber, getSegmentPath(segmentNumber).c_str());

  if (lseek(allfd[segmentNumber],
            rowInSegment * Offset(row_size), SEEK_SET)<0)
    PLERROR("problem in lseek: %s",strerror(errno));
  return allfd[segmentNumber];
}


template <class KT, class QR>
int SimpleDB<KT,QR>::seekToEnd() const
{

  if (max_records_file == 0)
    PLERROR("Attempting to seekToEnd without schema set");
        
  int last = lastSegment();
  int fd = allfd[last];
  if (fd == -1)
    PLERROR("Problem accessing database segment %d at path %s",
            last, getSegmentPath(last).c_str());

  off_t pos = lseek(fd, 0ul, SEEK_END);

  if (Offset(pos) / Offset(row_size) >= max_records_file)
    fd = seekToRow((last+1)*max_records_file);

  return fd;
}

    
template <class KT, class QR>
void SimpleDB<KT,QR>::openAllFiles() const
{
  closeAllFiles();                   
        
  int fd;
  fd = open(getSegmentPath(0).c_str(),
            access_mask | O_CREAT, 0777);
  if (fd == -1)
    PLERROR("Could not open main database segment %s: %s",
            getSegmentPath(0).c_str(), strerror(errno));
  allfd.push_back(fd);

  int index = 1;
  for ( ; ; ++index ) {
    fd = open(getSegmentPath(index).c_str(),
              access_mask);
    if (fd == -1)
      break;
    else
      allfd.push_back(fd);
  }
}    

    
template <class KT, class QR>
void SimpleDB<KT,QR>::closeAllFiles() const
{
  vector<int>::iterator it=allfd.begin(), end=allfd.end();
  for (; it != end; ++it)
    if (*it != -1) {
      close(*it);
    }
  allfd.clear();
}


template <class KT, class QR>
inline int SimpleDB<KT,QR>::lastSegment() const
{
  return (int)allfd.size() - 1;
}
    
    
template <class KT, class QR>
string SimpleDB<KT,QR>::getSegmentPath(int i) const
{
  string fullpath = path + name;
  if (i >= 1 && i <= 26) {
    string postfix(1, char('a'+i-1));
    fullpath += string("_") + postfix;
  }
  else if (i > 26)
    PLERROR("Too many segments in the database.");
  if(fullpath.find(".sdb")==string::npos)
    fullpath += ".sdb";
  return fullpath;
}
    

//#####  Indexing- and Query-Related Functions  ###########################

template <class KT, class QR>   
bool SimpleDB<KT,QR>::indexColumn(string column_name,
                                  string second_column)
{
  bool has_second_column = (second_column.size() > 0);
  int n,  start_pos,  column_precision =0,
    n2, start_pos2, column_precision2=0;
  if (!findColumn(column_name, n, start_pos, column_precision))
    return false;
  if (has_second_column &&
      !findColumn(second_column, n2, start_pos2, column_precision2))
    return false;

  RowNumber maxrows = size();
  RowNumber tablesize = RowNumber(table_size_multiplier*maxrows);
  if (maxrows <= 0 || tablesize <= 0) {
    PLWARNING("SimpleDB::indexColumn: cannot index a database of "
              "zero size.");
    return false;
  }
  if (!indexes[n]) {
    indexes[n] = new Index(tablesize, true);
    indexes[n]->initializeTable((unsigned int)tablesize);
  }
  Index& index = *indexes[n];
  index.resize(tablesize);
  index.flush();
    
  Row currow(&schema);
  IndexKey key(column_precision + column_precision2);
  typename IndexKey::iterator keybegin = key.begin();
  unsigned char* begin1 = currow.raw() + start_pos;
  unsigned char* end1   = begin1 + column_precision;
  unsigned char* begin2 = currow.raw() + start_pos2;
  unsigned char* end2   = begin2 + column_precision2;   
    
  for(RowNumber i=0; i<maxrows; ++i) {
    if (verbose && i % 1000000 == 0) {
      unsigned numclusters, maxcluster;
      index.diagnostics(numclusters, maxcluster);
      cerr << "indexing row " << i
           << "\t num. clusters=" << numclusters
           << "\t max. cluster size=" << maxcluster
           << endl;
    }

    getInRow(i,currow);
        
    copy(begin1, end1, keybegin);
    if (has_second_column)
      copy(begin2, end2, keybegin+column_precision-1);
    unsigned int addr = index.hashAddress(key);

    if (addr == Hash_UNUSED_TAG) {
      bool needresize = !index.add(key,QueryResult_t());
      if (needresize) {
        cerr << "Hash table unexpectedly full; exiting..." << endl;
        exit(1);
      }
      addr = index.hashAddress(key);
    }

    QueryResult_t* qr = index[addr];
    try {
      qr->push_back(i);
    }
    catch (logic_error& e) {
      cerr << "Exception caught during indexing: "
           << typeid(e).name() << endl
           << "Containing: " << e.what() << endl;
      throw;
    }   
  }
  return true;
}

template <class KT, class QR>   
void SimpleDB<KT,QR>::clearIndex(string column_name)
{
  int n, start_pos, column_precision=0;
  if (findColumn(column_name, n, start_pos, column_precision))
    indexes[n] = 0;
}
    
template <class KT, class QR>
QR SimpleDB<KT,QR>::findEqual(const unsigned char* lookfor,
                              string column_name, string second_column)
{
  int n, start_pos, column_precision;
  if (!findColumn(column_name, n, start_pos, column_precision))
    return EmptyResult;
  if (indexes[n])
    return findEqualIndexed(lookfor, column_name, second_column);
  else
    return findEqualLinear(lookfor, column_name, second_column);
}
    
    
template <class KT, class QR>   
const QR& SimpleDB<KT,QR>::findEqualIndexed(const unsigned char* lookfor,
                                            string column_name,
                                            string second_column)
{
  bool has_second_column = (second_column.size() > 0);
  int n,  start_pos,  column_precision =0,
    n2, start_pos2, column_precision2=0;
  if (!findColumn(column_name, n, start_pos, column_precision))
    return EmptyResult;
  if (has_second_column &&
      !findColumn(second_column, n2, start_pos2, column_precision2))
    return EmptyResult;

  if (!indexes[n])
    PLERROR("SimpleDB::indexColumn must be done before performing "
            "indexed searches on column %s", column_name.c_str());
        
  Index& index = *indexes[n];
  IndexKey key(lookfor, column_precision+column_precision2);
  unsigned int addr = index.hashAddress(key);
  if (addr == Hash_UNUSED_TAG)
    return EmptyResult;
  else
    return *index[addr];
}


template <class KT, class QR>
QR SimpleDB<KT,QR>::findEqualLinear(const unsigned char* lookfor,
                                    string column_name,
                                    string second_column)
{
  vuc lf(1);
  lf[0] = lookfor;
  return findEqualLinear(lf, column_name, second_column);
}
    
    
template <class KT, class QR>   
QR SimpleDB<KT,QR>::findEqualLinear(
  const vuc& lookfor,
  string column_name, string second_column)
{
  bool has_second_column = (second_column.size() > 0);
  int n,  start_pos,  column_precision =0,
    n2, start_pos2, column_precision2=0;
  if (!findColumn(column_name, n, start_pos, column_precision))
    return EmptyResult;
  if (has_second_column &&
      !findColumn(second_column, n2, start_pos2, column_precision2))
    return EmptyResult;

  QR qr;

  vector<IndexKey> key_lookfor(lookfor.size());
  vuc::const_iterator
    lookit, lookbeg = lookfor.begin(), lookend = lookfor.end();
  typename vector<IndexKey>::iterator
    keyit, keybeg = key_lookfor.begin(), keyend = key_lookfor.end();
  size_t len = column_precision+column_precision2;
        
  for (lookit=lookbeg, keyit=keybeg ; lookit != lookend ;
       ++lookit, ++keyit) {
    keyit->resize(len);
    copy(*lookit, *lookit+len, keyit->begin());
  }
        
  IndexKey key_dbrow(column_precision+column_precision2);
  typename IndexKey::iterator keybegin = key_dbrow.begin();

  RowNumber maxrows = size();
  Row currow(&schema);
  unsigned char* begin1 = currow.raw() + start_pos;
  unsigned char* end1   = begin1 + column_precision;
  unsigned char* begin2 = currow.raw() + start_pos2;
  unsigned char* end2   = begin2 + column_precision2;

  for (RowNumber i=0; i<maxrows; ++i) {
    if (verbose && i % 1000000 == 0) {
      cerr << "Searching row " << i << endl;
    }

    getInRow(i, currow);

    copy(begin1, end1, keybegin);
    if (has_second_column)
      copy(begin2, end2, keybegin+column_precision-1);

    for (keyit = keybeg ; keyit != keyend ; ++keyit) 
      if (*keyit == key_dbrow) {
        qr.push_back(i);
        if (verbose)
          cerr << "Found string \"" << *keyit
               << "\" at row " << i << endl;
      }
  }
        
  return qr;
}

#ifdef WIN32
#undef open
#undef close
#undef lseek
#undef read
#undef write
#undef unlink
#endif

} // end of namespace PLearn

#endif

---