hashtab.cpp

FastDb是高效的内存数据库系统

//-< HASHTAB.CPP >---------------------------------------------------*--------*
// FastDB                    Version 1.0         (c) 1999  GARRET    *     ?  *
// (Main Memory Database Management System)                          *   /\|  *
//                                                                   *  /  \  *
//                          Created:     20-Nov-98    K.A. Knizhnik  * / [] \ *
//                          Last update: 19-Dec-98    K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Extensible hash table implementation
//-------------------------------------------------------------------*--------*

#define INSIDE_FASTDB

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

BEGIN_FASTDB_NAMESPACE

int const dbHashTable::keySize[] = {
    1,  // tpBool
    1,  // tpInt1
    2,  // tpInt2
    4,  // tpInt4
    8,  // tpInt8
    4,  // tpReal4 
    8,  // tpReal8
    0,  // tpString,
    sizeof(oid_t), // tpReference
    -1, // tpArray,
    -1  // tpStructure,
};

oid_t dbHashTable::allocate(dbDatabase* db, size_t nRows)
{
    size_t size = dbInitHashTableSize;
    while (size <= nRows) { 
        size = (size+1)*2 - 1;
    }
    oid_t hashId = db->allocateObject(dbHashTableMarker);
    int nPages = (size+1) / dbIdsPerPage;
    oid_t pageId = db->allocateId(nPages);
    offs_t pos = db->allocate((size+1)*sizeof(oid_t));
    assert((pos & (dbPageSize-1)) == 0);
    memset(db->baseAddr+pos, 0, (size+1)*sizeof(oid_t));
    dbHashTable* hash = (dbHashTable*)db->get(hashId);
    hash->size = size;
    hash->page = pageId;
    hash->used = 0;
    while (--nPages >= 0) {
        db->currIndex[pageId++] = pos + dbPageObjectMarker;
        pos += dbPageSize;
    }
    return hashId;
}
 

inline unsigned dbHashTable::strHashCode(byte* key, int keylen) 
{
    unsigned h;
#ifdef IGNORE_CASE
    for (h = 0; --keylen >= 0;) { 
        int code = *key++;
        h = h*31 + toupper(code);
    }
#else
    for (h = 0; --keylen >= 0; h = h*31 + *key++);
#endif
    return h;
}

inline unsigned dbHashTable::hashCode(byte* key, int type, int keylen) 
{
    switch (type) { 
      case dbField::tpBool:
      case dbField::tpInt1:
        return *(db_nat1*)key;
      case dbField::tpInt2:
        return *(db_nat2*)key;
      case dbField::tpInt4:
      case dbField::tpReal4: 
        return *(db_nat4*)key;
      case dbField::tpInt8:
      case dbField::tpReal8: 
        return *(db_nat4*)key ^ *((db_nat4*)key+1);
      default:
      {  
          unsigned h;
          key += keylen;
          for (h = 0; --keylen >= 0; h = (h << 8) + *--key);
          return h;
      }
    }
}


inline int keycmp(void* k1, void* k2, int type, int keylen)
{
    switch (type) { 
      case dbField::tpBool:
      case dbField::tpInt1:
        return *(db_nat1*)k1 - *(db_nat1*)k2;
      case dbField::tpInt2:
        return *(db_nat2*)k1 - *(db_nat2*)k2;
      case dbField::tpInt4:
      case dbField::tpReal4: 
        return *(db_nat4*)k1 - *(db_nat4*)k2;
      case dbField::tpInt8:
      case dbField::tpReal8: 
        return *(db_nat8*)k1 == *(db_nat8*)k2 ? 0 : 1;
      default:
        return memcmp(k1, k2, keylen);
    }
}

void dbHashTable::insert(dbDatabase* db, oid_t hashId, 
                         oid_t rowId, int type, int sizeofType, int offs, size_t nRows)
{
    dbHashTable* hash = (dbHashTable*)db->get(hashId);
    byte* record = db->get(rowId);
    byte* key = record + offs;
    unsigned hashkey;
    if (type == dbField::tpString) { 
        int len = ((dbVarying*)key)->size - 1;
        key = record + ((dbVarying*)key)->offs;
        hashkey = strHashCode(key, len);
    } else {  
        hashkey = hashCode(key, type, sizeofType);
    }
    size_t size = hash->size;
    oid_t pageId = hash->page;
    if (size < nRows && hash->used >= size) {
        TRACE_MSG(("Reallocate hash table, used=%ld, size=%ld\n", hash->used, size));
        int nPages = (size+1) / dbIdsPerPage;
        size = (size+1)*2-1;
        oid_t newPageId = db->allocateId((size+1) / dbIdsPerPage);
        offs_t pos = db->allocate((size+1)*sizeof(oid_t));
        assert((pos & (dbPageSize-1)) == 0);
        memset(db->baseAddr + pos, 0, (size+1)*sizeof(oid_t));
        hash = (dbHashTable*)db->put(hashId);
        hash->size = size;
        hash->page = newPageId;
        size_t used = 0;
        while (--nPages >= 0) { 
            for (size_t i = 0; i < dbIdsPerPage; i++) { 
                oid_t itemId = ((oid_t*)db->get(pageId))[i];
                while (itemId != 0) { 
                    dbHashTableItem* item = (dbHashTableItem*)db->get(itemId);
                    oid_t nextId = item->next;
                    unsigned h = item->hash % size;
                    oid_t* tab = (oid_t*)(db->baseAddr + pos);
                    if (item->next != tab[h]) { 
                        item = (dbHashTableItem*)db->put(itemId);
                        tab = (oid_t*)(db->baseAddr + pos);
                        item->next = tab[h];
                    }
                    if (tab[h] == 0) { 
                        used += 1;
                    }
                    tab[h] = itemId;
                    itemId = nextId;
                }
            }
            db->freeObject(pageId++);
        }
        ((dbHashTable*)db->get(hashId))->used = used;
        pageId = newPageId;
        for (nPages = (size+1)/dbIdsPerPage; --nPages >= 0; pos += dbPageSize){
            db->currIndex[newPageId++] = pos + dbPageObjectMarker;
        }
    } 
    oid_t itemId = db->allocateObject(dbHashTableItemMarker);
    unsigned h = hashkey % size;
    oid_t* ptr = (oid_t*)db->put(pageId + h/dbIdsPerPage) + h%dbIdsPerPage;
    dbHashTableItem* item = (dbHashTableItem*)db->get(itemId);
    item->record = rowId;
    item->hash = hashkey;
    item->next = *ptr;
    *ptr = itemId;
    if (item->next == 0) { 
        ((dbHashTable*)db->get(hashId))->used += 1;
        db->file.markAsDirty(db->currIndex[hashId] & ~dbInternalObjectMarker, sizeof(dbHashTable));
    }
}

    
void dbHashTable::remove(dbDatabase* db, oid_t hashId, 
                         oid_t rowId, int type, int sizeofType, int offs)
{
    dbHashTable* hash = (dbHashTable*)db->get(hashId);
    byte* record = (byte*)db->getRow(rowId);
    byte* key = record + offs;
    unsigned hashkey;
    if (type == dbField::tpString) { 
        int len = ((dbVarying*)key)->size - 1;
        key = record + ((dbVarying*)key)->offs;
        hashkey = strHashCode(key, len);
    } else {  
        hashkey = hashCode(key, type, sizeofType);
    }
    unsigned h = hashkey % hash->size;
    oid_t pageId = hash->page + h / dbIdsPerPage;
    int i = h % dbIdsPerPage;
    oid_t itemId = ((oid_t*)db->get(pageId))[i];
    oid_t prevItemId = 0;
    while (true) { 
        assert(itemId != 0);
        dbHashTableItem* item = (dbHashTableItem*)db->get(itemId);
        if (item->record == rowId) { 
            oid_t next = item->next;
            if (prevItemId == 0) { 
                if (next == 0) { 
                    hash->used -= 1; // consistency can be violated
                    db->file.markAsDirty(db->currIndex[hashId] & ~dbInternalObjectMarker, 
                                         sizeof(dbHashTable));
                }
                *((oid_t*)db->put(pageId) + i) = next;
            } else {    
                item = (dbHashTableItem*)db->put(prevItemId);
                item->next = next;
            }
            db->freeObject(itemId);
            return;
        }
        prevItemId = itemId;
        itemId = item->next;
    }
}
    
void dbHashTable::find(dbDatabase* db, oid_t hashId, dbSearchContext& sc)
{
    dbHashTable* hash = (dbHashTable*)db->get(hashId);
    unsigned hashkey;
    unsigned keylen;
    if (hash->size == 0) { 
        return;
    }
    if (sc.type == dbField::tpString) { 
        keylen = strlen(sc.firstKey);
        hashkey = strHashCode((byte*)sc.firstKey, keylen);
    } else {  
        keylen = sc.sizeofType;
        hashkey = hashCode((byte*)sc.firstKey, sc.type, keylen);
    }
    unsigned h = hashkey % hash->size;
    oid_t itemId = 
        ((oid_t*)db->get(hash->page + h/dbIdsPerPage))[h % dbIdsPerPage];
    dbTable* table = (dbTable*)db->getRow(sc.cursor->table->tableId);
    while (itemId != 0) { 
        dbHashTableItem* item = (dbHashTableItem*)db->get(itemId);
        sc.probes += 1;
        if (item->hash == hashkey) { 
            byte* rec = (byte*)db->getRow(item->record);
            if ((sc.type == dbField::tpString
                 && keylen == ((dbVarying*)(rec + sc.offs))->size - 1
#ifdef IGNORE_CASE
                 && stricmp(sc.firstKey, 
                            (char*)rec+((dbVarying*)(rec+sc.offs))->offs) == 0)
#else
                 && keycmp(sc.firstKey, rec+((dbVarying*)(rec+sc.offs))->offs, 
                           sc.type, keylen) == 0)
#endif
                || (sc.type != dbField::tpString
                    && sc.comparator(sc.firstKey, rec + sc.offs, keylen) == 0)) 
            { 
                if (!sc.condition 
                    || db->evaluate(sc.condition, item->record, table, sc.cursor))
                {
                    if (!sc.cursor->add(item->record)) { 
                        return;
                    }
                }
            }
        }
        itemId = item->next;
    }
}



void dbHashTable::purge(dbDatabase* db, oid_t hashId)
{
    dbHashTable* hash = (dbHashTable*)db->put(hashId);
    oid_t pageId = hash->page;
    int nPages = (hash->size+1) / dbIdsPerPage;
    hash->used = 0;
    while (--nPages >= 0) { 
        for (size_t i = 0; i < dbIdsPerPage; i++) { 
            oid_t itemId = ((oid_t*)db->get(pageId))[i];
            while (itemId != 0) { 
                oid_t nextId = ((dbHashTableItem*)db->get(itemId))->next;
                db->freeObject(itemId);
                itemId = nextId;
            }
        }
        memset(db->put(pageId++), 0, dbPageSize);
    }
}

void dbHashTable::drop(dbDatabase* db, oid_t hashId)
{
    dbHashTable* hash = (dbHashTable*)db->get(hashId);
    oid_t pageId = hash->page;
    int nPages = (hash->size+1) / dbIdsPerPage;
    while (--nPages >= 0) { 
        for (size_t i = 0; i < dbIdsPerPage; i++) { 
            oid_t itemId = ((oid_t*)db->get(pageId))[i];
            while (itemId != 0) { 
                oid_t nextId = ((dbHashTableItem*)db->get(itemId))->next;
                db->freeObject(itemId);
                itemId = nextId;
            }
        }
        db->freeObject(pageId++);
    }
    db->freeObject(hashId);
}

END_FASTDB_NAMESPACE