localcli.cpp

来自「一个功能强大的内存数据库源代码,c++编写,有详细的注释」· C++ 代码 · 共 1,971 行 · 第 1/4 页

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

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, 0, transactionCommitDelay)) { 
	    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()) {
	    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()) {
	    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;
    }    
    for (stmt = s->stmts; stmt != NULL; stmt = next) {
	next = stmt->next;
	free_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 = new char[strlen(sql)+1];
    strcpy(stmt->sql, 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->n_autoincremented_columns = 0;
    stmt->oid = 0;
    stmt->next = s->stmts;
    stmt->updated = false;
    stmt->table = NULL;
    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) {
	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,
		     cli_column_get get)
{
    return cli_array_column_ex(statement, column_name, var_type, var_ptr, 
			       (cli_column_set_ex)set, (cli_column_get_ex)get);
}

int cli_array_column_ex(int               statement,
			char const*     column_name,
			int               var_type,
			void*             var_ptr,
			cli_column_set_ex set,
			cli_column_get_ex get)
{
    return dbCLI::instance.bind_array_column(statement, column_name, var_type, var_ptr, set, get);
}


int dbCLI::bind_array_column(int               statement,
			     char const*     column_name,
			     int               var_type,
			     void*             var_ptr,
			     cli_column_set_ex set,
			     cli_column_get_ex get)
{
    statement_desc* s = statements.get(statement);
    if (s == NULL) {
	return cli_bad_descriptor;
    }
    if (var_type < cli_asciiz || var_type > cli_array_of_string) {
	return cli_unsupported_type;
    }
    s->prepared = false;
    column_binding* cb = new column_binding;
    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 = NULL;
    cb->var_ptr = var_ptr;
    cb->set_fnc = set;
    cb->get_fnc = get;
    return cli_ok;
}

int dbCLI::match_columns(char const* table_name, statement_desc* stmt)
{    
    stmt->table = stmt->session->db->findTable(table_name);
    if (stmt->table == NULL) {
	return cli_table_not_found;
    }
    for (column_binding* cb = stmt->columns; cb != NULL; cb = cb->next) { 
	cb->field = stmt->table->find(cb->name);
	if (cb->field == NULL) { 
	    return cli_column_not_found;
	}
    }
    return cli_ok;
}

int cli_fetch(int statement, int for_update)
{
    return dbCLI::instance.fetch(statement, for_update);
}

int dbCLI::fetch(int statement, int for_update)
{
    statement_desc* stmt = statements.get(statement);    
    if (stmt == NULL) {
	return cli_bad_descriptor;
    }
    stmt->for_update = for_update;
    stmt->oid = 0;
    if (!stmt->prepared) {
	int tkn;
	sql_scanner scanner(stmt->sql);
	if (scanner.get() != tkn_select) {
	    return cli_bad_statement;
	}
	if ((tkn = scanner.get()) == tkn_all) {
	    tkn = scanner.get();
	}
	if (tkn == tkn_from && scanner.get() == tkn_ident) {
	    int rc = match_columns(scanner.identifier(), stmt);
	    if (rc != cli_ok) { 
		return rc;
	    }
	} else { 
	    return cli_bad_statement;
	}
	char* p = scanner.current_position(), *q = p;
	parameter_binding* pb = stmt->params;
	stmt->query.reset();
	while (*p != '\0') {
	    if (*p == '\'') {
		do {
		    do {
			p += 1;
		    } while (*p != '\0' && *p != '\'');
		    if (*p == '\0') {
			return cli_bad_statement;
		    }
		} while (*++p == '\'');
	    } else if (*p == '%') {
		if (p != q) { 
		    *p = '\0';
		    stmt->query.append(dbQueryElement::qExpression, q);