cursor.cpp

最新版本！fastdb是高效的内存数据库系统

//-< 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 "rtree.h"
#include <ctype.h>

BEGIN_FASTDB_NAMESPACE

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) const
{
    for (segment *seg = first; seg != NULL; seg = seg->next) { 
        for (int i = 0, n = seg->nRows; i < n; i++) {
            *oids++ = seg->rows[i];
        }
    }
}

void dbSelection::truncate(size_t from, size_t length)
{
    if (from == 0 && length >= nRows) { 
        // do nothing
        return;
    }
    segment *src = NULL;
    bool empty = true;
    if (from < nRows) { 
        for (src = first; src != NULL; src = src->next) { 
            if (from < src->nRows) { 
                empty = false;
                break;
            }
            from -= src->nRows;
        }
    }
    if (from + length > nRows) { 
        length = nRows - from;
    }
    nRows = 0;
    segment* dst = first;
    size_t pos = 0;
    if (!empty) { 
        while (length != 0) { 
            size_t n = src->nRows - from;
            if (n > length) { 
                n = length;
            }
            if (dst->nRows == pos) { 
                dst = dst->next;
                pos = 0;
            }
            if (n > dst->nRows - pos) { 
                n = dst->nRows - pos;
            }
            memcpy(dst->rows + pos, src->rows + from, n*sizeof(oid_t));
            pos += n;
            length -= n;
            nRows += n;
            if ((from += n) == src->nRows) { 
                if ((src = src->next) == NULL) { 
                    break;
                }
                from = 0;
            }
        }
    }
    dst->nRows = pos;
    segment* next = dst->next;
    dst->next = NULL;
    while (next != NULL) { 
        dst = next;
        next = dst->next; 
        delete dst;
    }
}

int dbSelection::compare(oid_t o1, oid_t o2, dbOrderByNode* order)
{
    dbDatabase* db = order->table->db;
    char* p = (char*)db->getRow(o1);
    char* q = (char*)db->getRow(o2);
    int diff = 0;
    do { 
        if (order->expr != NULL) { 
            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;
            iattr1.oid = o1;
            iattr2.record = sattr2.base = (byte*)q;
            iattr2.oid = o2;
            db->execute(order->expr, iattr1, sattr1);
            db->execute(order->expr, iattr2, sattr2);
            switch (order->expr->type) { 
              case tpInteger:
                diff = sattr1.ivalue < sattr2.ivalue ? -1 : sattr1.ivalue == sattr2.ivalue ? 0 : 1;
                break;
              case tpReal:
                diff = sattr1.fvalue < sattr2.fvalue ? -1 : sattr1.fvalue == sattr2.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 < sattr2.oid ? -1 : sattr1.oid == sattr2.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;
}

#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 && checkForDuplicatedIsEnabled && 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);
    }
}

void dbAnyCursor::deallocateBitmap() 
{
    if (bitmap != NULL) { 
        delete[] bitmap;
        bitmapSize = 0;
        bitmap = NULL;
    }
}

oid_t* dbAnyCursor::toArrayOfOid(oid_t* arr) const
{ 
    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) 
{ 
    if (ref.oid != 0) { 
        db->handleError(dbDatabase::NullReferenceError, "Attempt to set NULL reference as cursor current value");
    } 
    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.probes = 0;
    sc.offs = field->dbsOffs;
    sc.cursor = this;
    sc.condition = NULL;
    sc.prefixLength = 0;
    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 (gotoFirst() && prefetch) {
        fetch();