cursor.cpp

fastdb-2.92的源码

//-< 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) const
{
    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) 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) 
{ 
    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;