ttree.cpp

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

//-< TTREE.CPP >-----------------------------------------------------*--------*
// FastDB                    Version 1.0         (c) 1999  GARRET    *     ?  *
// (Main Memory Database Management System)                          *   /\|  *
//                                                                   *  /  \  *
//                          Created:     20-Nov-98    K.A. Knizhnik  * / [] \ *
//                          Last update:  6-Jan-99    K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// T-Tree implementation
//-------------------------------------------------------------------*--------*

#define INSIDE_FASTDB

#include "fastdb.h"
#include "ttree.h"

#ifdef USE_LOCALE_SETTINGS
#ifdef IGNORE_CASE
#define strcmp(x, y) stricoll(x, y)
#else
#define strcmp(x, y) strcoll(x, y)
#endif
#else
#ifdef IGNORE_CASE
#define strcmp(x, y) stricmp(x, y)
#endif
#endif

inline int keycmp(void* p, void* q, int type, int sizeofType, dbUDTComparator comparator)
{
  switch (type)
  {

  case dbField::tpBool:
    return *(bool*)p - *(bool*)q;

  case dbField::tpInt1:
    return *(int1*)p - *(int1*)q;

  case dbField::tpInt2:
    return *(int2*)p - *(int2*)q;

  case dbField::tpInt4:
    return *(int4*)p < *(int4*)q ? -1 : *(int4*)p == *(int4*)q ? 0 : 1;

  case dbField::tpInt8:
    return *(db_int8*)p < *(db_int8*)q ? -1 : *(db_int8*)p == *(db_int8*)q ? 0 : 1;

  case dbField::tpReference:
    return *(oid_t*)p < *(oid_t*)q ? -1 : *(oid_t*)p == *(oid_t*)q ? 0 : 1;

  case dbField::tpReal4:
    return *(real4*)p < *(real4*)q ? -1 : *(real4*)p == *(real4*)q ? 0 : 1;

  case dbField::tpReal8:
    return *(real8*)p < *(real8*)q ? -1 : *(real8*)p == *(real8*)q ? 0 : 1;

  case dbField::tpRawBinary:
    return comparator(p, q, sizeofType);

  default:
    assert(false);
    return 0;
  }
}

void dbTtree::find(dbDatabase* db, oid_t treeId, dbSearchContext& sc)
{

  oid_t rootId = ((dbTtree*)db->get
                  (treeId))->root;

  if (rootId != 0)
  {

    ((dbTtreeNode*)db->get
     (rootId))->find(db, sc);
  }
}

oid_t dbTtree::allocate(dbDatabase* db)
{
  oid_t oid = db->allocateObject(dbTtreeMarker);

  dbTtree* tree = (dbTtree*)db->get
                  (oid);

  tree->root = 0;

  return oid;
}

void dbTtree::insert(dbDatabase* db, oid_t treeId, oid_t recordId,
                     int type, int sizeofType, dbUDTComparator comparator, int offs)
{

  oid_t rootId = ((dbTtree*)db->get
                  (treeId))->root;

  if (rootId == 0)
  {
    oid_t nodeId = dbTtreeNode::allocate(db, recordId);
    ((dbTtree*)db->put(treeId))->root = nodeId;
  }
  else
  {
    byte* rec = (byte*)db->getRow(recordId);
    byte* key = rec + offs;

    if (type == dbField::tpString)
    {
      key = rec + ((dbVarying*)key)->offs;
    }

    oid_t nodeId = rootId;
    dbTtreeNode::insert(db, nodeId, recordId, key, type, sizeofType, comparator, offs);

    if (rootId != nodeId)
    {
      ((dbTtree*)db->put(treeId))->root = nodeId;
    }
  }
}


void dbTtree::remove
  (dbDatabase* db, oid_t treeId, oid_t recordId,
   int type, int sizeofType, dbUDTComparator comparator, int offs)
{

  oid_t rootId = ((dbTtree*)db->get
                  (treeId))->root;

  byte* rec = (byte*)db->getRow(recordId);

  byte* key = rec + offs;

  if (type == dbField::tpString)
  {
    key = rec + ((dbVarying*)key)->offs;
  }

  oid_t nodeId = rootId;

  int h = dbTtreeNode::remove
            (db, nodeId, recordId, key, type, sizeofType, comparator, offs);

  assert(h >= 0);

  if (nodeId != rootId)
  {
    ((dbTtree*)db->put(treeId))->root = nodeId;
  }
}


void dbTtree::purge(dbDatabase* db, oid_t treeId)
{

  oid_t rootId = ((dbTtree*)db->get
                  (treeId))->root;

  dbTtreeNode::purge(db, rootId);

  ((dbTtree*)db->put(treeId))->root = 0;
}

void dbTtree::drop(dbDatabase* db, oid_t treeId)
{
  purge(db, treeId);
  db->freeObject(treeId);
}


void dbTtree::traverseForward(dbDatabase* db, oid_t treeId,
                              dbAnyCursor* cursor)
{

  oid_t rootId = ((dbTtree*)db->get
                  (treeId))->root;

  if (rootId != 0)
  {

    ((dbTtreeNode*)db->get
     (rootId))->traverseForward(db, cursor);
  }
}

void dbTtree::traverseBackward(dbDatabase* db, oid_t treeId,
                               dbAnyCursor* cursor)
{

  oid_t rootId = ((dbTtree*)db->get
                  (treeId))->root;

  if (rootId != 0)
  {

    ((dbTtreeNode*)db->get
     (rootId))->traverseBackward(db, cursor);
  }
}


void dbTtree::traverseForward(dbDatabase* db, oid_t treeId,
                              dbAnyCursor* cursor, dbExprNode* condition)
{

  oid_t rootId = ((dbTtree*)db->get
                  (treeId))->root;

  if (rootId != 0)
  {

    ((dbTtreeNode*)db->get
     (rootId))->traverseForward(db, cursor,condition);
  }
}


void dbTtree::traverseBackward(dbDatabase* db, oid_t treeId,
                               dbAnyCursor* cursor, dbExprNode* condition)
{

  oid_t rootId = ((dbTtree*)db->get
                  (treeId))->root;

  if (rootId != 0)
  {

    ((dbTtreeNode*)db->get
     (rootId))->traverseBackward(db,cursor,condition);
  }
}




bool dbTtreeNode::find(dbDatabase* db, dbSearchContext& sc)
{
  char* rec;
  int   diff;
  int   l, r, m, n = nItems;

  sc.probes += 1;
  dbTable* table = (dbTable*)db->getRow(sc.cursor->table->tableId);

  if (sc.type == dbField::tpString)
  {
    if (sc.firstKey != NULL)
    {
      rec = (char*)db->getRow(item[0]);
      diff = strcmp(sc.firstKey, rec+((dbVarying*)(rec+sc.offs))->offs);

      if (diff >= sc.firstKeyInclusion)
      {
        rec = (char*)db->getRow(item[n-1]);
        diff = strcmp(sc.firstKey,
                      rec+((dbVarying*)(rec+sc.offs))->offs);

        if (diff >= sc.firstKeyInclusion)
        {
          if (right != 0)
          {

            return ((dbTtreeNode*)db->get
                    (right))->find(db, sc);
          }

          return true;
        }

        for (l = 0, r = n; l < r;)
        {
          m = (l + r) >> 1;
          rec = (char*)db->getRow(item[m]);
          diff = strcmp(sc.firstKey,
                        rec + ((dbVarying*)(rec+sc.offs))->offs);

          if (diff >= sc.firstKeyInclusion)
          {
            l = m+1;
          }
          else
          {
            r = m;
          }
        }

        while (r < n)
        {
          rec = (char*)db->getRow(item[r]);

          if (sc.lastKey != NULL
              && strcmp(rec + ((dbVarying*)(rec+sc.offs))->offs,
                        sc.lastKey) >= sc.lastKeyInclusion)
          {
            return false;
          }

          if (!sc.condition
              || db->evaluate(sc.condition, item[r], table, sc.cursor))
          {
            if (!sc.cursor->add
                (item[r]))
            {
              return false;
            }
          }

          r += 1;
        }

        if (right != 0)
        {

          return ((dbTtreeNode*)db->get
                  (right))->find(db, sc);
        }

        return true;
      }
    }

    if (left != 0)
    {
      if (!((dbTtreeNode*)db->get
            (left))->find(db, sc))
      {
        return false;
      }
    }

    for (l = 0; l < n; l++)
    {
      rec = (char*)db->getRow(item[l]);

      if (sc.lastKey != NULL
          && strcmp(rec + ((dbVarying*)(rec+sc.offs))->offs,
                    sc.lastKey) >= sc.lastKeyInclusion)
      {
        return false;
      }

      if (!sc.condition || db->evaluate(sc.condition, item[l], table, sc.cursor))
      {
        if (!sc.cursor->add
            (item[l]))
        {
          return false;
        }
      }
    }

    if (right != 0)
    {

      return ((dbTtreeNode*)db->get
              (right))->find(db, sc);
    }
  }
  else
  {
    if (sc.firstKey != NULL)
    {
      rec = (char*)db->getRow(item[0]);
      diff = keycmp(sc.firstKey, rec+sc.offs, sc.type, sc.sizeofType, sc.comparator);

      if (diff >= sc.firstKeyInclusion)
      {
        rec = (char*)db->getRow(item[n-1]);
        diff = keycmp(sc.firstKey, rec+sc.offs, sc.type, sc.sizeofType, sc.comparator);

        if (diff >= sc.firstKeyInclusion)
        {
          if (right != 0)
          {

            return ((dbTtreeNode*)db->get
                    (right))->find(db, sc);
          }

          return true;
        }

        for (l = 0, r = n; l < r;)
        {
          m = (l + r) >> 1;
          rec = (char*)db->getRow(item[m]);
          diff = keycmp(sc.firstKey, rec+sc.offs, sc.type, sc.sizeofType, sc.comparator);

          if (diff >= sc.firstKeyInclusion)
          {
            l = m+1;
          }
          else
          {
            r = m;
          }
        }

        while (r < n)
        {
          rec = (char*)db->getRow(item[r]);

          if (sc.lastKey != NULL
              && keycmp(rec+sc.offs, sc.lastKey, sc.type, sc.sizeofType, sc.comparator)
              >= sc.lastKeyInclusion)
          {
            return false;
          }

          if (!sc.condition
              || db->evaluate(sc.condition, item[r], table, sc.cursor))
          {
            if (!sc.cursor->add
                (item[r]))
            {
              return false;
            }
          }

          r += 1;
        }

        if (right != 0)
        {

          return ((dbTtreeNode*)db->get
                  (right))->find(db, sc);
        }

        return true;
      }
    }

    if (left != 0)
    {
      if (!((dbTtreeNode*)db->get
            (left))->find(db, sc))
      {
        return false;
      }
    }

    for (l = 0; l < n; l++)
    {
      rec = (char*)db->getRow(item[l]);

      if (sc.lastKey != NULL && keycmp(rec+sc.offs, sc.lastKey, sc.type, sc.sizeofType, sc.comparator)
          >= sc.lastKeyInclusion)
      {
        return false;
      }

      if (!sc.condition || db->evaluate(sc.condition, item[l], table, sc.cursor))
      {
        if (!sc.cursor->add
            (item[l]))
        {
          return false;
        }
      }
    }

    if (right != 0)
    {

      return ((dbTtreeNode*)db->get
              (right))->find(db, sc);
    }
  }

  return true;
}


oid_t dbTtreeNode::allocate(dbDatabase* db, oid_t recordId)
{
  oid_t nodeId = db->allocateObject(dbTtreeNodeMarker);

  dbTtreeNode* node = (dbTtreeNode*)db->get
                      (nodeId);

  node->nItems = 1;

  node->item[0] = recordId;

  node->left = node->right = 0;

  node->balance = 0;

  return nodeId;
}

bool dbTtreeNode::insert(dbDatabase* db, oid_t& nodeId, oid_t recordId,
                         void* key, int type, int sizeofType, dbUDTComparator comparator, int offs)
{

  dbTtreeNode* node = (dbTtreeNode*)db->get
                      (nodeId);

  char* rec = (char*)db->getRow(node->item[0]);

  int n = node->nItems;

  int diff = (type == dbField::tpString)
             ? strcmp((char*)key, rec + ((dbVarying*)(rec+offs))->offs)
             : keycmp(key, rec+offs, type, sizeofType, comparator);

  if (diff <= 0)
  {
    oid_t leftId = node->left;

    if ((leftId == 0 || diff == 0) && node->nItems != pageSize)
    {
      node = (dbTtreeNode*)db->put(nodeId);

      for (int i = n; i > 0; i--)
        node->item[i] = node->item[i-1];

      node->item[0] = recordId;

      node->nItems += 1;

      return false;
    }

    if (leftId == 0)
    {
      leftId = allocate(db, recordId);
      node = (dbTtreeNode*)db->put(nodeId);
      node->left = leftId;
    }
    else
    {
      oid_t childId = leftId;
      bool grow = insert(db, childId, recordId, key, type, sizeofType, comparator, offs);

      if (childId != leftId)
      {
        ((dbTtreeNode*)db->put(nodeId))->left = leftId = childId;
      }

      if (!grow)
        return false;
    }

    node = (dbTtreeNode*)db->put(nodeId);

    if (node->balance > 0)
    {
      node->balance = 0;
      return false;
    }
    else if (node->balance == 0)
    {
      node->balance = -1;
      return true;
    }
    else
    {
      dbTtreeNode* left = (dbTtreeNode*)db->put(leftId);

      node = (dbTtreeNode*)db->get
             (nodeId);

      if (left->balance < 0)
      { // single LL turn
        node->left = left->right;
        left->right = nodeId;
        node->balance = 0;
        left->balance = 0;
        nodeId = leftId;
      }
      else
      { // double LR turn
        oid_t rightId = left->right;
        dbTtreeNode* right = (dbTtreeNode*)db->put(rightId);

        left = (dbTtreeNode*)db->get
               (leftId);

        node = (dbTtreeNode*)db->get
               (nodeId);

        left->right = right->left;

        right->left = leftId;

        node->left = right->right;

        right->right = nodeId;

        node->balance = (right->balance < 0) ? 1 : 0;

        left->balance = (right->balance > 0) ? -1 : 0;

        right->balance = 0;

        nodeId = rightId;
      }

      return false;
    }
  }

  rec = (char*)db->getRow(node->item[n-1]);
  diff = (type == dbField::tpString)
         ? strcmp((char*)key, rec + ((dbVarying*)(rec+offs))->offs)
         : keycmp(key, rec+offs, type, sizeofType, comparator);

  if (diff >= 0)
  {
    oid_t rightId = node->right;

    if ((rightId == 0 || diff == 0) && node->nItems != pageSize)
    {
      node = (dbTtreeNode*)db->put(nodeId);
      node->item[n] = recordId;
      node->nItems += 1;
      return false;
    }

    if (rightId == 0)
    {
      rightId = allocate(db, recordId);
      node = (dbTtreeNode*)db->put(nodeId);
      node->right = rightId;
    }
    else
    {
      oid_t childId = rightId;
      bool grow = insert(db, childId, recordId, key, type, sizeofType, comparator, offs);

      if (childId != rightId)
      {
        ((dbTtreeNode*)db->put(nodeId))->right = rightId = childId;
      }

      if (!grow)
        return false;
    }

    node = (dbTtreeNode*)db->put(nodeId);

    if (node->balance < 0)
    {
      node->balance = 0;
      return false;
    }
    else if (node->balance == 0)
    {
      node->balance = 1;
      return true;
    }
    else
    {
      dbTtreeNode* right = (dbTtreeNode*)db->put(rightId);