localcli.cpp

来自「FastDb是高效的内存数据库系统」· C++ 代码 · 共 3,024 行 · 第 1/5 页

//-< 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);
}

session_desc* cli_get_session(int id)
{
  return dbCLI::instance.get_session(id);
}

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;
    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.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)
  {
    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;