localcli.cpp

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

//-< LOCALCLI.CPP >--------------------------------------------------*--------*
// FastDB                    Version 1.0         (c) 1999  GARRET    *     ?  *
// (Main Memory Database Management System)                          *   /\|  *
//                                                                   *  /  \  *
//                          Created:     20-Jun-2002  K.A. Knizhnik  * / [] \ *
//                          Last update: 20-Jun-2002  K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Implementation of local C interface to database
//-------------------------------------------------------------------*--------*

#define INSIDE_FASTDB

#include "localcli.h"
#include "ttree.h"
#include "rtree.h"
#include "hashtab.h"
#include "symtab.h"
#include <ctype.h>

USE_FASTDB_NAMESPACE

dbCLI dbCLI::instance;

int cli_open(char const* server_url,
             int         max_connect_attempts,
             int         reconnect_timeout_sec)
{
    return cli_bad_address;
}

int cli_create(char const* databaseName,
               char const* filePath,
               unsigned    transactionCommitDelay, 
               int         openAttr, 
               size_t      initDatabaseSize,
               size_t      extensionQuantum,
               size_t      initIndexSize,
               size_t      fileSizeLimit)
{
    return dbCLI::instance.create_session(databaseName, filePath, transactionCommitDelay, openAttr, 
                                          initDatabaseSize, extensionQuantum, initIndexSize, fileSizeLimit);
}

int dbCLI::create_session(char const* databaseName, 
                          char const* filePath,
                          unsigned    transactionCommitDelay, 
                          int         openAttr, 
                          size_t      initDatabaseSize,
                          size_t      extensionQuantum,
                          size_t      initIndexSize,
                          size_t      fileSizeLimit)
{
    dbCriticalSection cs(sessionMutex);
    dbDatabase* db = NULL; 
    session_desc* s;
    for (s = active_session_list; s != NULL; s = s->next) { 
        if (strcmp(s->name, databaseName) == 0) { 
            db = s->db;
            db->accessCount += 1;
            break;
        }
    }
    if (db == NULL) { 
        db = new dbDatabase((openAttr & cli_open_readonly) 
                            ? (openAttr & cli_open_concurrent) 
                              ? dbDatabase::dbConcurrentRead : dbDatabase::dbReadOnly 
                            : (openAttr & cli_open_concurrent) 
                              ? dbDatabase::dbConcurrentUpdate : dbDatabase::dbAllAccess,
                            initDatabaseSize, 
                            extensionQuantum, 
                            initIndexSize);     
        if (!db->open(databaseName, filePath, INFINITE, transactionCommitDelay)) {
            db->close();
            delete db;
            return cli_database_not_found;
        }
        db->setFileSizeLimit(fileSizeLimit);
        dbTable* table = (dbTable*)db->getRow(dbMetaTableId);
        dbTableDescriptor* metatable = new dbTableDescriptor(table);
        db->linkTable(metatable, dbMetaTableId);
        oid_t tableId = table->firstRow;
        while (tableId != 0) {
            table = (dbTable*)db->getRow(tableId);
            dbTableDescriptor* desc;
            for (desc = db->tables; desc != NULL && desc->tableId != tableId; desc = desc->nextDbTable);
            if (desc == NULL) { 
                desc = new dbTableDescriptor(table);
                db->linkTable(desc, tableId);
                desc->setFlags();
            }
            tableId = table->next;
        }
        if (!db->completeDescriptorsInitialization()) {
            db->close();
            delete db;
            return cli_table_not_found;
        }
        db->accessCount = 1;
    }
    s = sessions.allocate();
    s->name = new char[strlen(databaseName) + 1];
    strcpy(s->name, databaseName);
    s->db = db;
    s->stmts = NULL;
    s->next = active_session_list;
    s->existed_tables = db->tables;
    s->dropped_tables = NULL;
    active_session_list = s;
    return s->id;
}

int cli_create_replication_node(int         nodeId,
                                int         nServers,
                                char*       nodeNames[],
                                char const* databaseName, 
                                char const* filePath, 
                                int         openAttr, 
                                size_t      initDatabaseSize,
                                size_t      extensionQuantum,
                                size_t      initIndexSize,
                                size_t      fileSizeLimit)
{
    return dbCLI::instance.create_replication_node(nodeId,
                                                   nServers,
                                                   nodeNames,
                                                   databaseName, 
                                                   filePath, 
                                                   openAttr, 
                                                   initDatabaseSize,
                                                   extensionQuantum,
                                                   initIndexSize,
                                                   fileSizeLimit);
}

int dbCLI::create_replication_node(int         nodeId,
                                   int         nServers,
                                   char*       nodeNames[],
                                   char const* databaseName, 
                                   char const* filePath, 
                                   int         openAttr, 
                                   size_t      initDatabaseSize,
                                   size_t      extensionQuantum,
                                   size_t      initIndexSize,
                                   size_t      fileSizeLimit)
{
#ifdef REPLICATION_SUPPORT
    dbCriticalSection cs(sessionMutex);
    dbDatabase* db = NULL; 
    session_desc* s;
    for (s = active_session_list; s != NULL; s = s->next) { 
        if (strcmp(s->name, databaseName) == 0) { 
            db = s->db;
            db->accessCount += 1;
            break;
        }
    }
    if (db == NULL) { 
        db = new dbReplicatedDatabase((openAttr & cli_open_readonly) 
                                      ? (openAttr & cli_open_concurrent) 
                                      ? dbDatabase::dbConcurrentRead : dbDatabase::dbReadOnly 
                                      : (openAttr & cli_open_concurrent) 
                                      ? dbDatabase::dbConcurrentUpdate : dbDatabase::dbAllAccess,
                                      initDatabaseSize, 
                                      extensionQuantum, 
                                      initIndexSize);            
        if (!((dbReplicatedDatabase*)db)->open(databaseName, filePath, nodeId, nodeNames, nServers)) {
            return cli_database_not_found;
        }
        db->setFileSizeLimit(fileSizeLimit);
        dbTable* table = (dbTable*)db->getRow(dbMetaTableId);
        dbTableDescriptor* metatable = new dbTableDescriptor(table);
        db->linkTable(metatable, dbMetaTableId);
        oid_t tableId = table->firstRow;
        while (tableId != 0) {
            table = (dbTable*)db->getRow(tableId);
            dbTableDescriptor* desc;
            for (desc = db->tables; desc != NULL && desc->tableId != tableId; desc = desc->nextDbTable);
            if (desc == NULL) { 
                desc = new dbTableDescriptor(table);
                db->linkTable(desc, tableId);
                desc->setFlags();
            }
            tableId = table->next;
        }
        if (!db->completeDescriptorsInitialization()) {
            db->close();
            delete db;
            return cli_table_not_found;
        }
        db->accessCount = 1;
    }
    s = sessions.allocate();
    s->name = new char[strlen(databaseName) + 1];
    strcpy(s->name, databaseName);
    s->db = db;
    s->stmts = NULL;
    s->next = active_session_list;
    s->existed_tables = db->tables;
    s->dropped_tables = NULL;
    active_session_list = s;
    return s->id;
#else 
    return cli_not_implemented;
#endif
}


int cli_close(int session)
{
    return dbCLI::instance.close(session);
}

int dbCLI::close(int session)
{
    dbCriticalSection cs(sessionMutex);
    statement_desc *stmt, *next;
    session_desc* s = sessions.get(session);
    if (s == NULL) {
        return cli_bad_descriptor;
    }    
    dbCriticalSection cs2(s->mutex);
    for (stmt = s->stmts; stmt != NULL; stmt = next) {
        next = stmt->next;
        release_statement(stmt);
    }
    if (--s->db->accessCount == 0) { 
        dbTableDescriptor *desc, *next_desc;
        for (desc = s->db->tables; desc != NULL; desc = next_desc) {
            next_desc = desc->nextDbTable;
            if (!desc->isStatic) { 
                delete desc;
            }
        }
        s->db->tables = NULL;
        s->db->close();
        delete s->db;
    }
    while (s->dropped_tables != NULL) {
        dbTableDescriptor* next = s->dropped_tables->nextDbTable;
        delete s->dropped_tables;
        s->dropped_tables = next;
    }
    session_desc** spp;
    for (spp = &active_session_list; *spp != s; spp = &(*spp)->next);
    *spp = s->next;
    delete[] s->name;
    sessions.free(s);

    return cli_ok;
}

int cli_statement(int session, char const* sql)
{
    return dbCLI::instance.create_statement(session, sql);
}

int dbCLI::create_statement(int session, char const* sql)
{
    session_desc* s = sessions.get(session);
    if (s == NULL) {
        return cli_bad_descriptor;
    }
    statement_desc* stmt = statements.allocate();
    stmt->sql.put(strlen(sql)+1);
    strcpy(stmt->sql.base(), sql);
    stmt->columns = NULL;
    stmt->params = NULL;
    stmt->session = s;
    stmt->for_update = 0;
    stmt->first_fetch = true;
    stmt->prepared = false;
    stmt->n_params = 0;
    stmt->n_columns = 0;
    stmt->record_struct = NULL;
    stmt->n_autoincremented_columns = 0;
    stmt->oid = 0;
    stmt->updated = false;
    stmt->table = NULL;
    {
        dbCriticalSection cs(s->mutex);
        stmt->next = s->stmts;
        s->stmts = stmt;
    }
    char const* p = sql;
    parameter_binding** last = &stmt->params;
    while (*p != '\0') {
        if (*p == '\'') {
            do {
                do {
                    p += 1;
                } while (*p != '\0' && *p != '\'');
                if (*p == '\0') {
                    *last = NULL;
                    free_statement(stmt);
                    return cli_bad_statement;
                }
            } while (*++p == '\'');
        } else if (*p == '%') {
            stmt->n_params += 1;
            char const* q = p++;
            while (isalnum((unsigned char)*p) || *p == '_') p += 1;
            if (*p == '%') {
                *last = NULL;
                free_statement(stmt);
                return cli_bad_statement;
            }
            parameter_binding* pb = parameter_allocator.allocate();
            int len = p - q;
            pb->name = new char[len+1];
            memcpy(pb->name, q, len);
            pb->name[len] = '\0';
            *last = pb;
            last = &pb->next;
            pb->var_ptr = NULL;
        } else {
            p += 1;
        }
    }
    *last = NULL;
    return stmt->id;
}

int cli_parameter(int           statement,
                  char const* param_name,
                  int           var_type,
                  void*         var_ptr)
{
    return dbCLI::instance.bind_parameter(statement, param_name, var_type, var_ptr);
}


int dbCLI::bind_parameter(int           statement,
                          char const* param_name,
                          int           var_type,
                          void*         var_ptr)
{
    if ((unsigned)var_type >= cli_array_of_oid && var_type != cli_rectangle) {
        return cli_unsupported_type;
    }
    statement_desc* s = statements.get(statement);
    if (s == NULL) {
        return cli_bad_descriptor;
    }
    s->prepared = false;
    for (parameter_binding* pb = s->params; pb != NULL; pb = pb->next) {
        if (strcmp(pb->name, param_name) == 0) {
            pb->var_ptr  = var_ptr;
            pb->var_type = var_type;
            return cli_ok;
        }
    }
    return cli_parameter_not_found;
}

int cli_column(int           statement,
               char const* column_name,
               int           var_type,
               int*          var_len,
               void*         var_ptr)
{
    return dbCLI::instance.bind_column(statement, column_name, var_type, var_len, var_ptr);
}

int dbCLI::bind_column(int           statement,
                       char const* column_name,
                       int           var_type,
                       int*          var_len,
                       void*         var_ptr)
{
    statement_desc* s = statements.get(statement);
    if (s == NULL) {
        return cli_bad_descriptor;
    }
    if ((unsigned)var_type >= cli_unknown) { 
        return cli_unsupported_type;
    }
    s->prepared = false;
    if (var_type == cli_autoincrement) {
        s->n_autoincremented_columns += 1;
    }
    column_binding* cb = column_allocator.allocate();
    cb->name = new char[strlen(column_name) + 1];
    cb->next = s->columns;
    s->columns = cb;
    s->n_columns += 1;
    strcpy(cb->name, column_name);
    cb->var_type = var_type;
    cb->var_len = var_len;
    cb->var_ptr = var_ptr;
    cb->set_fnc = NULL;
    cb->get_fnc = NULL;
    return cli_ok;
}

int cli_array_column(int            statement,
                     char const*  column_name,
                     int            var_type,
                     void*          var_ptr,
                     cli_column_set set,