cursor.cpp

来自「FastDb是高效的内存数据库系统」· C++ 代码 · 共 1,098 行 · 第 1/2 页

//-< CURSOR.CPP >----------------------------------------------------*--------*
// FastDB                    Version 1.0         (c) 1999  GARRET    *     ?  *
// (Main Memory Database Management System)                          *   /\|  *
//                                                                   *  /  \  *
//                          Created:     20-Nov-98    K.A. Knizhnik  * / [] \ *
//                          Last update: 21-Dec-98    K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Table cursor
//-------------------------------------------------------------------*--------*

#define INSIDE_FASTDB

#include "fastdb.h"
#include "compiler.h"
#include "hashtab.h"
#include "ttree.h"
#include <ctype.h>


dbSelection::segment*  dbSelection::createNewSegment(dbSelection::segment* after)
{
  return new segment(after);
}

inline void dbSelection::reset()
{
  for (segment *next, *seg = first; seg != NULL; seg = next)
  {
    next = seg->next;
    delete seg;
  }

  first = last = curr = NULL;
  nRows = 0;
  pos = 0;
}

void dbSelection::reverse()
{
  segment *next, *seg;

  for (seg = first; seg != NULL; seg = next)
  {
    next = seg->next;
    seg->next = seg->prev;
    seg->prev = next;

    for (int l = 0, r = seg->nRows-1; l < r; l++, r--)
    {
      oid_t oid = seg->rows[l];
      seg->rows[l] = seg->rows[r];
      seg->rows[r] = oid;
    }
  }

  seg = first;
  first = last;
  last = seg;
}

void dbSelection::toArray(oid_t* oids)
{
  for (segment *seg = first; seg != NULL; seg = seg->next)
  {
    for (int i = 0, n = seg->nRows; i < n; i++)
    {
      *oids++ = seg->rows[i];
    }
  }
}


int dbSelection::compare(dbRecord* a, dbRecord* b, dbOrderByNode* order)
{
  char* p = (char*)a;
  char* q = (char*)b;
  int diff = 0;

  do
  {
    if (order->expr != NULL)
    {
      dbDatabase* db = order->table->db;
      dbInheritedAttribute   iattr1;
      dbInheritedAttribute   iattr2;
      dbSynthesizedAttribute sattr1;
      dbSynthesizedAttribute sattr2;
      iattr1.db = iattr2.db = db;
      iattr1.table = iattr2.table = (dbTable*)db->getRow(order->table->tableId);
      iattr1.record = sattr1.base = (byte*)p;
      iattr2.record = sattr2.base = (byte*)q;
      db->execute(order->expr, iattr1, sattr1);
      db->execute(order->expr, iattr2, sattr2);

      switch (order->expr->type)
      {

      case tpInteger:
        diff = sattr1.ivalue < sattr1.ivalue ? -1 : sattr1.ivalue == sattr1.ivalue ? 0 : 1;
        break;

      case tpReal:
        diff = sattr1.fvalue < sattr1.fvalue ? -1 : sattr1.fvalue == sattr1.fvalue ? 0 : 1;
        break;

      case tpBoolean:
diff = sattr1.bvalue != 0 ? sattr2.bvalue != 0 ? 0 : 1 : sattr2.bvalue != 0 ? -1 : 0;
        break;

      case tpString:
#ifdef USE_LOCALE_SETTINGS

        diff = strcoll((char*)sattr1.array.base, (char*)sattr2.array.base);
#else

        diff = strcmp((char*)sattr1.array.base, (char*)sattr2.array.base);
#endif

        break;

      case tpReference:
        diff = sattr1.oid < sattr1.oid ? -1 : sattr1.oid == sattr1.oid ? 0 : 1;
        break;

      default:
        assert(false);
      }
    }
    else
    {
      int offs = order->field->dbsOffs;

      switch (order->field->type)
      {

      case dbField::tpBool:
        diff = *(bool*)(p + offs) - *(bool*)(q + offs);
        break;

      case dbField::tpInt1:
        diff = *(int1*)(p + offs) - *(int1*)(q + offs);
        break;

      case dbField::tpInt2:
        diff = *(int2*)(p + offs) - *(int2*)(q + offs);
        break;

      case dbField::tpInt4:

      case dbField::tpArray:
        diff = *(int4*)(p + offs) < *(int4*)(q + offs) ? -1
               : *(int4*)(p + offs) == *(int4*)(q + offs) ? 0 : 1;
        break;

      case dbField::tpInt8:
        diff = *(db_int8*)(p + offs) < *(db_int8*)(q + offs) ? -1 :
               *(db_int8*)(p + offs) == *(db_int8*)(q + offs) ? 0 : 1;
        break;

      case dbField::tpReference:
        diff = *(oid_t*)(p + offs) < *(oid_t*)(q + offs) ? -1 :
               *(oid_t*)(p + offs) == *(oid_t*)(q + offs) ? 0 : 1;
        break;

      case dbField::tpReal4:
        diff = *(real4*)(p + offs) < *(real4*)(q + offs) ? -1 :
               *(real4*)(p + offs) == *(real4*)(q + offs) ? 0 : 1;
        break;

      case dbField::tpReal8:
        diff = *(real8*)(p + offs) < *(real8*)(q + offs) ? -1 :
               *(real8*)(p + offs) == *(real8*)(q + offs) ? 0 : 1;
        break;

      case dbField::tpString:
#ifdef USE_LOCALE_SETTINGS

        diff = strcoll(p + ((dbVarying*)(p + offs))->offs,
                       q + ((dbVarying*)(q + offs))->offs);
#else

        diff = strcmp(p + ((dbVarying*)(p + offs))->offs,
                      q + ((dbVarying*)(q + offs))->offs);
#endif

        break;

      case dbField::tpRawBinary:
        diff = order->field->comparator(p + offs, q + offs, order->field->dbsSize);
        break;

      default:
        assert(false);
      }
    }

    if (!order->ascent)
    {
      diff = -diff;
    }
  }
  while (diff == 0 && (order = order->next) != NULL);

  return diff;
}

int dbSelection::compare(oid_t a, oid_t b, dbOrderByNode* order)
{
  dbDatabase* db = order->table->db;
  return compare(db->getRow(a), db->getRow(b), order);
}

#ifdef USE_HEAP_SORT

#define ELEM(i)   index[(i-1)/quantum]->rows[(i-1)%quantum]
#define ROW(i)    db->getRow(ELEM(i))
#define SWAP(i,j) temp = ELEM(i), ELEM(i) = ELEM(j), ELEM(j) = temp

void dbSelection::sort(dbDatabase* db, dbOrderByNode* order)
{
  size_t i, j, k, n = nRows;
  oid_t temp;

  if (n <= 1)
  {
    return;
  }

  TRACE_MSG(("Sort %d records\n", n));
  segment** index = new segment*[(n + quantum - 1) / quantum];
  segment* seg = first;

  for (i = 0; seg != NULL; seg = seg->next)
  {
    index[i++] = seg;
  }

  for (i = n/2, j = i; i >= 1; i--)
  {
    k = i;
    oid_t topId = ELEM(k);
    dbRecord* top = db->getRow(topId);

    do
    {
      if (k*2 == n || compare(ROW(k*2), ROW(k*2+1), order) > 0)
      {
        if (compare(top, ROW(k*2), order) >= 0)
        {
          break;
        }

        ELEM(k) = ELEM(k*2);
        k = k*2;
      }
      else
      {
        if (compare(top, ROW(k*2+1), order) >= 0)
        {
          break;
        }

        ELEM(k) = ELEM(k*2+1);
        k = k*2+1;
      }
    }
    while (k <= j);

    ELEM(k) = topId;
  }

  for (i = n; i >= 2; i--)
  {
    SWAP(1, i);
    oid_t topId = ELEM(1);
    dbRecord* top = db->getRow(topId);

    for (k = 1, j = (i-1)/2; k <= j;)
    {
      if (k*2 == i-1 || compare(ROW(k*2), ROW(k*2+1), order) > 0)
      {
        if (compare(top, ROW(k*2), order) >= 0)
        {
          break;
        }

        ELEM(k) = ELEM(k*2);
        k = k*2;
      }
      else
      {
        if (compare(top, ROW(k*2+1), order) >= 0)
        {
          break;
        }

        ELEM(k) = ELEM(k*2+1);
        k = k*2+1;
      }
    }

    ELEM(k) = topId;
  }

  delete[] index;
}

#else

struct dbSortContext
{
  dbOrderByNode* order;
};

static dbThreadContext<dbSortContext> sortThreadContext;

#ifdef USE_STDLIB_QSORT

static int compareRecords(void const* a, void const* b)
{
  dbSortContext* ctx = sortThreadContext.get();
  return dbSelection::compare(*(oid_t*)a, *(oid_t*)b, ctx->order);
}


void dbSelection::sort(dbDatabase* db, dbOrderByNode* order)
{
  size_t n = nRows;

  if (n <= 1)
  {
    return;
  }

  TRACE_MSG(("Sort %d records\n", n));
  oid_t* oids = new oid_t[n];
  toArray(oids);
  dbSortContext ctx;
  ctx.order = order;
  sortThreadContext.set(&ctx);
  qsort(oids, n, sizeof(oid_t), &compareRecords);
  oid_t* p = oids;

  for (segment *seg = first; seg != NULL; seg = seg->next)
  {
    for (int i = 0, n = seg->nRows; i < n; i++)
    {
      seg->rows[i] = *p++;
    }
  }
}

#else

#include "iqsort.h"

struct ObjectRef
{
  oid_t oid;

  bool operator > (ObjectRef& ref)
  {
    return compare(this, &ref) > 0;
  }

  bool operator < (ObjectRef& ref)
  {
    return compare(this, &ref) < 0;
  }

  bool operator >= (ObjectRef& ref)
  {
    return compare(this, &ref) >= 0;
  }

  bool operator <= (ObjectRef& ref)
  {
    return compare(this, &ref) <= 0;
  }

  bool operator == (ObjectRef& ref)
  {
    return compare(this, &ref) == 0;
  }

  bool operator != (ObjectRef& ref)
  {
    return compare(this, &ref) != 0;
  }

  static int compare(ObjectRef* a, ObjectRef* b)
  {
    dbSortContext* ctx = sortThreadContext.get();
    return dbSelection::compare(a->oid, b->oid, ctx->order);
  }
};


void dbSelection::sort(dbDatabase* db, dbOrderByNode* order)
{
  size_t n = nRows;

  if (n <= 1)
  {
    return;
  }

  TRACE_MSG(("Sort %d records\n", n));
  ObjectRef* refs = new ObjectRef[n];
  segment *seg;
  int k = 0;

  for (seg = first; seg != NULL; seg = seg->next)
  {
    for (int i = 0, n = seg->nRows; i < n; i++)
    {
      refs[k++].oid = seg->rows[i];
    }
  }

  dbSortContext ctx;
  ctx.order = order;
  sortThreadContext.set(&ctx);
  iqsort(refs, n);
  k = 0;

  for (seg = first; seg != NULL; seg = seg->next)
  {
    for (int i = 0, n = seg->nRows; i < n; i++)
    {
      seg->rows[i] = refs[k++].oid;
    }
  }

  delete[] refs;
}

#endif
#endif

void dbAnyCursor::checkForDuplicates()
{
  if (!eliminateDuplicates && limit > 1)
  {
    eliminateDuplicates = true;
    size_t size = (db->currIndexSize + 31) / 32;

    if (size > bitmapSize)
    {
      delete[] bitmap;
      bitmap = new int4[size];
      bitmapSize = size;
    }

    memset(bitmap, 0, size*4);
  }
}


oid_t* dbAnyCursor::toArrayOfOid(oid_t* arr)
{
  if (arr == NULL)
  {
    arr = new oid_t[selection.nRows];
  }

  if (allRecords)
  {
    oid_t* oids = arr;

    for (oid_t oid = firstId; oid != 0; oid = db->getRow(oid)->next)
    {
      *oids++ = oid;
    }
  }
  else
  {
    selection.toArray(arr);
  }

  return arr;
}

void dbAnyCursor::setCurrent(dbAnyReference const& ref)
{
  removed = false;
  assert(ref.oid != 0);
  reset();
  db->beginTransaction(type == dbCursorForUpdate
                       ? dbDatabase::dbExclusiveLock : dbDatabase::dbSharedLock);
  db->threadContext.get()->cursors.link(this);
  currId = ref.oid;

  add
    (currId);

  if (prefetch)
  {
    fetch();
  }
}


int dbAnyCursor::selectByKey(char const* key, void const* value)
{
  dbFieldDescriptor* field = table->find(key);
  assert(field != NULL);
  assert(field->hashTable != 0 || field->tTree != 0);
  reset();
  db->beginTransaction(type == dbCursorForUpdate
                       ? dbDatabase::dbExclusiveLock : dbDatabase::dbSharedLock);
  db->threadContext.get()->cursors.link(this);
  dbSearchContext sc;
  sc.db = db;
  sc.offs = field->dbsOffs;
  sc.cursor = this;
  sc.condition = NULL;
  sc.firstKey = sc.lastKey = (char*)value;
  sc.firstKeyInclusion = sc.lastKeyInclusion = true;
  sc.comparator = field->comparator;
  sc.sizeofType = field->dbsSize;
  sc.type = field->type;

  if (field->hashTable != 0)
  {
    dbHashTable::find(db, field->hashTable, sc);
  }
  else
  {
    dbTtree::find(db, field->tTree, sc);
  }

  if (prefetch && gotoFirst())
  {
    fetch();
  }

  return selection.nRows;
}

int dbAnyCursor::seek(oid_t oid)
{
  int pos = 0;

  if (gotoFirst())
  {
    do
    {
      if (currId == oid)