src/server/DatabaseManagerImpl.cxx

   1 /***************************************************************************
   2  *   Copyright (C) 2007 by www.databasecache.com                           *
   3  *   Contact: praba_tuty@databasecache.com                                 *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) any later version.                                   *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15   ***************************************************************************/
  16 #include<Database.h>
  17 #include<DatabaseManager.h>
  18 #include<DatabaseManagerImpl.h>
  19 #include<os.h>
  20 #include<Table.h>
  21 #include<TableImpl.h>
  22 #include<Transaction.h>
  23 #include<CatalogTables.h>
  24 #include<Index.h>
  25 #include<Lock.h>
  26 #include<Debug.h>
  27 #include<Config.h>
  28 #include<Process.h>
  29
  30
  31 DatabaseManagerImpl::~DatabaseManagerImpl()
  32 {
  33     //Note:Databases are closed by the session interface
  34     delete tMgr_;
  35     delete lMgr_;
  36 }
  37
  38 void DatabaseManagerImpl::createLockManager()
  39 {
  40     lMgr_ = new LockManager(systemDatabase_);
  41     return;
  42 }
  43
  44 void DatabaseManagerImpl::createTransactionManager()
  45 {
  46
  47     tMgr_ = new TransactionManager();
  48     tMgr_->setFirstTrans(systemDatabase_->getSystemDatabaseTrans(0));
  49     return;
  50 }
  51
  52 DbRetVal DatabaseManagerImpl::openSystemDatabase()
  53 {
  54     DbRetVal rv = openDatabase(SYSTEMDB);
  55     if (rv != OK) return rv;
  56     systemDatabase_ = db_;
  57     db_ = NULL;
  58     if (NULL == systemDatabase_)
  59     {
  60         printError(ErrAlready, "Database is already opened");
  61         return ErrAlready;
  62     }
  63     printDebug(DM_Database, "Opened system database");
  64     logFinest(logger, "Opened system database");
  65     return OK;
  66 }
  67
  68 DbRetVal DatabaseManagerImpl::closeSystemDatabase()
  69 {
  70     Database *db  = db_;
  71     //make them to point to system database file descriptor
  72     //and database pointer
  73     db_ = systemDatabase_;
  74     closeDatabase();
  75     db_ = db;
  76     printDebug(DM_Database, "Closed system database");
  77     logFinest(logger, "Closed System database");
  78     return OK;
  79 }
  80
  81 DbRetVal DatabaseManagerImpl::createDatabase(const char *name, size_t size)
  82 {
  83     if (NULL != db_ )
  84     {
  85         printError(ErrAlready, "Database is already created");
  86         return ErrAlready;
  87     }
  88     caddr_t rtnAddr = (caddr_t) NULL;
  89     shared_memory_id shm_id = 0;
  90
  91     char *startaddr = (char*)Conf::config.getMapAddress();
  92     shared_memory_key key = 0;
  93     if (0 == strcmp(name, SYSTEMDB))
  94     {
  95
  96         key = Conf::config.getSysDbKey();
  97     }
  98     else
  99     {
 100         startaddr = startaddr + Conf::config.getMaxSysDbSize();
 101         key = Conf::config.getUserDbKey();
 102     }
 103     shm_id = os::shm_create(key, size, 0666);
 104     if (-1 == shm_id)
 105     {
 106         printError(ErrOS, "Shared memory create failed");
 107         return ErrOS;
 108     }
 109
 110     void *shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 111     rtnAddr  = (caddr_t) shm_ptr;
 112     if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
 113     {
 114         printError(ErrOS, "Shared memory attach returned -ve value %d", rtnAddr);
 115         return ErrOS;
 116     }
 117     memset(shm_ptr, 0, size );
 118     db_ = new Database();
 119     printDebug(DM_Database, "Creating database:%s",name);
 120
 121     //TODO:for user database do not have transtable and processtable mutex
 122     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 123     db_->setDatabaseID(1);
 124     db_->setName(name);
 125     db_->setMaxSize(size);
 126     db_->setMaxChunks(MAX_CHUNKS);
 127     db_->initAllocDatabaseMutex();
 128     db_->initTransTableMutex();
 129     db_->initDatabaseMutex();
 130     db_->initProcessTableMutex();
 131
 132     //compute the first page after book keeping information
 133     size_t offset = os::alignLong(sizeof (DatabaseMetaData));
 134     //Only for system db chunk array, trans array and proc array will be there
 135     if (0 == strcmp(name, SYSTEMDB))
 136     {
 137         offset = offset + os::alignLong( MAX_CHUNKS  * sizeof (Chunk));
 138         offset = offset + os::alignLong( Conf::config.getMaxTrans()   * sizeof(Transaction));
 139         offset = offset + os::alignLong( Conf::config.getMaxProcs() * sizeof(ThreadInfo));
 140     }
 141     int multiple = os::floor(offset / PAGE_SIZE);
 142     char *curPage = (((char*)rtnAddr) + ((multiple + 1) * PAGE_SIZE));
 143
 144     db_->setCurrentPage(curPage);
 145     db_->setFirstPage(curPage);
 146
 147     if (0 == strcmp(name, SYSTEMDB)) return OK;
 148
 149     //Allocate new chunk to store hash index nodes
 150     Chunk *chunkInfo = createUserChunk(sizeof(HashIndexNode));
 151     if (NULL == chunkInfo)
 152     {
 153         printError(ErrSysInternal, "Failed to allocate hash index nodes chunk");
 154         return ErrSysInternal;
 155     }
 156     printDebug(DM_Database, "Creating Chunk for storing Hash index nodes %x",
 157                                                                   chunkInfo);
 158
 159     db_->setHashIndexChunk(chunkInfo);
 160     logFinest(logger, "Created database %s" , name);
 161
 162     return OK;
 163 }
 164
 165 DbRetVal DatabaseManagerImpl::deleteDatabase(const char *name)
 166 {
 167     shared_memory_id shm_id = 0;
 168     if (0 == strcmp(name, SYSTEMDB))
 169     {
 170         shm_id = os::shm_open(Conf::config.getSysDbKey(), 100, 0666);
 171         os::shmctl(shm_id, IPC_RMID);
 172     } else {
 173         shm_id = os::shm_open(Conf::config.getUserDbKey(), 100, 0666);
 174         os::shmctl(shm_id, IPC_RMID);
 175     }
 176     logFinest(logger, "Deleted database %s" , name);
 177     return OK;
 178 }
 179
 180 DbRetVal DatabaseManagerImpl::openDatabase(const char *name)
 181 {
 182     long size = Conf::config.getMaxSysDbSize();
 183     char *startaddr = (char*)Conf::config.getMapAddress();
 184     if (0 == strcmp(name , SYSTEMDB))
 185     {
 186         if (NULL !=systemDatabase_)
 187         {
 188             printError(ErrAlready, "System Database already open");
 189             return ErrAlready;
 190         }
 191     }
 192     else
 193     {
 194         if (NULL ==systemDatabase_)
 195         {
 196             printError(ErrNotOpen, "System Database not open");
 197             return ErrNotOpen;
 198         }
 199         size = Conf::config.getMaxDbSize();
 200         startaddr = startaddr +  Conf::config.getMaxSysDbSize();
 201     }
 202     if (NULL != db_)
 203     {
 204         printError(ErrAlready, "User Database already open");
 205         return ErrAlready;
 206     }
 207     //system db should be opened before user database files
 208     caddr_t rtnAddr = (caddr_t) NULL;
 209
 210     shared_memory_id shm_id = 0;
 211     shared_memory_key key = 0;
 212
 213     if (0 == strcmp(name, SYSTEMDB))
 214         key = Conf::config.getSysDbKey();
 215     else
 216        key = Conf::config.getUserDbKey();
 217
 218
 219     int ret = ProcessManager::mutex.getLock(false);
 220     //If you are not getting lock ret !=0, it means somebody else is there.
 221     //he will close the database.
 222     if (ret != 0)
 223     {
 224         printError(ErrSysInternal, "Another thread calling open:Wait and then Retry\n");
 225         return ErrSysInternal;
 226     }
 227     void *shm_ptr = NULL;
 228     bool firstThread = false;
 229     //printf("PRABA::DEBUG:: opendb %d %s\n", ProcessManager::noThreads, name);
 230     if (ProcessManager::noThreads == 0 && 0 == strcmp(name, SYSTEMDB)
 231        || ProcessManager::noThreads == 1 && 0 != strcmp(name, SYSTEMDB) ) {
 232        shm_id = os::shm_open(key, size, 0666);
 233        if (shm_id == -1 )
 234        {
 235            printError(ErrOS, "Shared memory open failed");
 236            ProcessManager::mutex.releaseLock(false);
 237            return ErrOS;
 238        }
 239        shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 240        if (0 == strcmp(name, SYSTEMDB))
 241        {
 242              firstThread = true;
 243              ProcessManager::sysAddr = (char*) shm_ptr;
 244        }
 245        else
 246        {
 247               ProcessManager::usrAddr = (char*) shm_ptr;
 248        }
 249     } else {
 250         if (0 == strcmp(name, SYSTEMDB))
 251               shm_ptr = ProcessManager::sysAddr;
 252         else
 253               shm_ptr = ProcessManager::usrAddr;
 254     }
 255     ProcessManager::mutex.releaseLock(false);
 256
 257
 258     rtnAddr  = (caddr_t) shm_ptr;
 259
 260     if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
 261     {
 262         printError(ErrOS, "Shared memory attach returned -ve value %x %d", shm_ptr, errno);
 263         return ErrOS;
 264     }
 265     db_ = new Database();
 266     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 267
 268     if (firstThread) ProcessManager::systemDatabase = db_;
 269
 270     printDebug(DM_Database, "Opening database: %s", name);
 271     logFinest(logger, "Opened database %s" , name);
 272     return OK;
 273 }
 274
 275 DbRetVal DatabaseManagerImpl::closeDatabase()
 276 {
 277
 278     if (NULL == db_)
 279     {
 280         //Database is already closed
 281         return OK;
 282     }
 283     printDebug(DM_Database, "Closing database: %s",(char*)db_->getName());
 284     //check if this is the last thread to be deregistered
 285     int ret = ProcessManager::mutex.getLock(false);
 286     //If you are not getting lock ret !=0, it means somebody else is there.
 287     //he will close the database.
 288     if (ret == 0) {
 289     //printf("PRABA::FOR DEBUG closedb %d %s\n", ProcessManager::noThreads, (char*)db_->getName());
 290     if (ProcessManager::noThreads == 0 && 0 == strcmp((char*)db_->getName(), SYSTEMDB)
 291        || ProcessManager::noThreads == 1 && 0 != strcmp((char*)db_->getName(),  SYSTEMDB) )  {
 292                 os::shm_detach((char*)db_->getMetaDataPtr());
 293             }
 294     }
 295     ProcessManager::mutex.releaseLock(false);
 296     logFinest(logger, "Closed database");
 297     delete db_;
 298     db_ = NULL;
 299     return OK;
 300 }
 301 //Assumes that system database mutex is taken before calling this.
 302 Chunk* DatabaseManagerImpl::createUserChunk(size_t size)
 303 {
 304     //Allocate new node in system database to store
 305     Chunk *chunk = getSystemTableChunk(UserChunkTableId);
 306     void *ptr = chunk->allocate(systemDatabase_);
 307     if (NULL == ptr)
 308     {
 309          printError(ErrNoMemory, "Allocation failed for User chunk catalog table");
 310          return NULL;
 311     }
 312     Chunk *chunkInfo = (Chunk*)ptr;
 313     chunkInfo->initMutex();
 314     if (0 != size) chunkInfo->setSize(size);
 315     if (chunkInfo->allocSize_ > PAGE_SIZE)
 316         chunkInfo->curPage_ = db_->getFreePage(chunkInfo->allocSize_);
 317     else
 318         chunkInfo->curPage_ = db_->getFreePage();
 319     if ( NULL == chunkInfo->curPage_)
 320     {
 321         chunkInfo->destroyMutex();
 322         chunk->free(db_, ptr);
 323         printError(ErrNoMemory, "Database full: No space to allocate from database");
 324         return NULL;
 325     }
 326     PageInfo* firstPageInfo = ((PageInfo*)chunkInfo->curPage_);
 327     if (chunkInfo->allocSize_ > PAGE_SIZE)
 328     {
 329         int multiple = os::floor(chunkInfo->allocSize_ / PAGE_SIZE);
 330         int offset = ((multiple + 1) * PAGE_SIZE);
 331         firstPageInfo->setPageAsUsed(offset);
 332     }
 333     else
 334         firstPageInfo->setPageAsUsed(chunkInfo->allocSize_);
 335     if (0 == size)
 336     {
 337         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
 338         varInfo->isUsed_ = 0;
 339         varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
 340
 341     }
 342     chunkInfo->firstPage_ = chunkInfo->curPage_;
 343
 344     if (0 == size)
 345         chunkInfo->setAllocType(VariableSizeAllocator);
 346     else
 347         chunkInfo->setAllocType(FixedSizeAllocator);
 348
 349     //TODO::Generate chunkid::use tableid
 350     chunkInfo->setChunkID(-1);
 351     printDebug(DM_Database, "Creating new User chunk chunkID:%d size: %d firstPage:%x",
 352                                -1, chunkInfo->allocSize_, firstPageInfo);
 353
 354     return chunkInfo;
 355 }
 356
 357
 358 //Assumes that system database mutex is taken before calling this.
 359 DbRetVal DatabaseManagerImpl::deleteUserChunk(Chunk *chunk)
 360 {
 361     //Go to the pages and set them to notUsed
 362     Page *page = chunk->firstPage_;
 363     PageInfo* pageInfo = ((PageInfo*)page);
 364     //Here...sure that atleast one page will be there even no tuples
 365     //are inserted.so not checking if pageInfo == NULL
 366     while( pageInfo->nextPage_ != NULL)
 367     {
 368         PageInfo *prev = pageInfo;
 369         pageInfo = (PageInfo*)(pageInfo->nextPage_);
 370         //sets pageInfo->isUsed_ = 0 and pageInfo->hasFreeSpace_ = 0
 371         //and initializes the page content to zero
 372         if(NULL == pageInfo->nextPageAfterMerge_)
 373             os::memset(prev, 0, PAGE_SIZE);
 374         else
 375         {
 376             int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 377             os::memset(prev, 0, size);
 378         }
 379         printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, prev);
 380     }
 381     //The above loop wont execute for the last page
 382     //and for the case where table has only one page
 383     if(NULL == pageInfo->nextPageAfterMerge_)
 384         os::memset(pageInfo, 0, PAGE_SIZE);
 385     else
 386     {
 387         int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 388         os::memset(pageInfo, 0, size);
 389     }
 390     printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, pageInfo);
 391     chunk->chunkID_ = -1;
 392     chunk->allocSize_ = 0;
 393     chunk->curPage_ = NULL;
 394     chunk->firstPage_ = NULL;
 395     chunk->destroyMutex();
 396     printDebug(DM_Database,"deleting user chunk:%x",chunk);
 397     return OK;
 398 }
 399
 400 //-1 -> Unable to create chunk. No memory
 401 //-2 -> Unable to update the catalog tables
 402 DbRetVal DatabaseManagerImpl::createTable(const char *name, TableDef &def)
 403 {
 404     DbRetVal rv = OK;
 405     int fldCount = def.getFieldCount();
 406     //If total field count is less than 32, then 1 integer is used to store all null
 407     //information, if it is more then 1 char is used to store null information
 408     //of each field
 409     //This is to done to reduce cpu cycles for small tables
 410     int addSize = 0;
 411     if (fldCount > 31) addSize = 4; else addSize = os::align(fldCount);
 412     size_t sizeofTuple = os::align(def.getTupleSize())+addSize;
 413     rv = systemDatabase_->getDatabaseMutex();
 414     if (OK != rv ) {
 415         printError(rv, "Unable to get Database mutex");
 416         return rv;
 417     }
 418
 419     void *tptr =NULL;
 420     void *chunk = NULL;
 421
 422     //check whether table already exists
 423     CatalogTableTABLE cTable(systemDatabase_);
 424     cTable.getChunkAndTblPtr(name, chunk, tptr);
 425     if (NULL != tptr)
 426     {
 427         systemDatabase_->releaseDatabaseMutex();
 428         printError(ErrAlready, "Table %s already exists", name);
 429         return ErrAlready;
 430     }
 431
 432     //create a chunk to store the tuples
 433     Chunk *ptr = createUserChunk(sizeofTuple);
 434     if (NULL  == ptr)
 435     {
 436         systemDatabase_->releaseDatabaseMutex();
 437         printError(ErrNoResource, "Unable to create user chunk");
 438         return ErrNoResource;
 439     }
 440     printDebug(DM_Database,"Created UserChunk:%x", ptr);
 441
 442     //add row to TABLE
 443     int tblID = ((Chunk*)ptr)->getChunkID();
 444     rv = cTable.insert(name, tblID, sizeofTuple,
 445                                    def.getFieldCount(), ptr, tptr);
 446     if (OK != rv)
 447     {
 448         deleteUserChunk(ptr);
 449         systemDatabase_->releaseDatabaseMutex();
 450         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 451         return ErrSysInternal;
 452     }
 453     printDebug(DM_Database,"Inserted into TABLE:%s",name);
 454     //add rows to FIELD
 455     FieldIterator iter = def.getFieldIterator();
 456     CatalogTableFIELD cField(systemDatabase_);
 457     rv = cField.insert(iter, tblID ,tptr);
 458     if (OK != rv)
 459     {
 460         deleteUserChunk(ptr);
 461         void *cptr, *ttptr;//Dummy as remove below needs both these OUT params
 462         cTable.remove(name, cptr, ttptr);
 463         systemDatabase_->releaseDatabaseMutex();
 464         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 465         return ErrSysInternal;
 466     }
 467     printDebug(DM_Database,"Inserted into FIELD:%s",name);
 468     systemDatabase_->releaseDatabaseMutex();
 469     printDebug(DM_Database,"Table Created:%s",name);
 470     logFinest(logger, "Table Created %s" , name);
 471     return OK;
 472 }
 473
 474 //TODO::If any operation fails in between, then we may have some
 475 //dangling tuples, say we have have rows in INDEX table
 476 //which will not have any corresponding entries in TABLE
 477 //CHANGE the sequence so that it deletes from the bottom as
 478 //opposed to start from top as is written now
 479 DbRetVal DatabaseManagerImpl::dropTable(const char *name)
 480 {
 481     void *chunk = NULL;
 482     void *tptr =NULL;
 483     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 484     if (OK != rv) {
 485         printError(ErrSysInternal, "Unable to get database mutex");
 486         return ErrSysInternal;
 487     }
 488
 489     //remove the entry in TABLE
 490     CatalogTableTABLE cTable(systemDatabase_);
 491     rv = cTable.remove(name, chunk, tptr);
 492     if (OK != rv) {
 493         systemDatabase_->releaseDatabaseMutex();
 494         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 495         return ErrSysInternal;
 496     }
 497     printDebug(DM_Database,"Deleted from TABLE:%s",name);
 498
 499     //remove the entries in the FIELD table
 500     CatalogTableFIELD cField(systemDatabase_);
 501     rv = cField.remove(tptr);
 502     if (OK != rv) {
 503         systemDatabase_->releaseDatabaseMutex();
 504         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 505         return ErrSysInternal;
 506     }
 507     printDebug(DM_Database,"Deleted from FIELD:%s",name);
 508
 509     rv = deleteUserChunk((Chunk*)chunk);
 510     if (OK != rv) {
 511         systemDatabase_->releaseDatabaseMutex();
 512         printError(rv, "Unable to delete the chunk");
 513         return rv;
 514     }
 515     printDebug(DM_Database,"Deleted UserChunk:%x", chunk);
 516
 517     //TODO::check whether indexes are available and drop that also.
 518     CatalogTableINDEX cIndex(systemDatabase_);
 519     int noIndexes = cIndex.getNumIndexes(tptr);
 520     char *idxName;
 521     for (int i =1 ; i<= noIndexes; i++) {
 522         idxName = cIndex.getIndexName(tptr, i);
 523         dropIndexInt(idxName, false);
 524     }
 525     Chunk *chunkNode = systemDatabase_->getSystemDatabaseChunk(UserChunkTableId);
 526     chunkNode->free(systemDatabase_, (Chunk *) chunk);
 527     systemDatabase_->releaseDatabaseMutex();
 528     printDebug(DM_Database, "Deleted Table %s" , name);
 529     logFinest(logger, "Deleted Table %s" , name);
 530     return OK;
 531 }
 532
 533 //Return values: NULL for table not found
 534 Table* DatabaseManagerImpl::openTable(const char *name)
 535 {
 536     DbRetVal ret = OK;
 537     //TODO::store table handles in list so that if it is
 538     //not closed by the application. destructor shall close it.
 539     TableImpl *table = new TableImpl();
 540     table->setDB(db_);
 541     table->setSystemDB(systemDatabase_);
 542     table->setLockManager(lMgr_);
 543     table->setTrans(&(tMgr_->trans));
 544
 545     //to store the chunk pointer of table
 546     void *chunk = NULL;
 547
 548     //to store the tuple pointer of the table
 549     void *tptr =NULL;
 550
 551     //TODO::need to take shared lock on the table so that
 552     //all ddl operation will be denied on that table
 553     //which includes index creation, alter table
 554
 555     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 556     if (OK != rv) {
 557         printError(ErrSysInternal, "Unable to get database mutex");
 558         return NULL;
 559     }
 560     CatalogTableTABLE cTable(systemDatabase_);
 561     ret = cTable.getChunkAndTblPtr(name, chunk, tptr);
 562     if ( OK != ret)
 563     {
 564         printError(ErrNotExists, "Table not exists %s", name);
 565         return NULL;
 566     }
 567     TABLE *tTuple = (TABLE*)tptr;
 568     table->setTableInfo(tTuple->tblName_, tTuple->tblID_, tTuple->length_,
 569                         tTuple->numFlds_, tTuple->numIndexes_, tTuple->chunkPtr_);
 570
 571     //get field information from FIELD table
 572     CatalogTableFIELD cField(systemDatabase_);
 573     cField.getFieldInfo(tptr, table->fldList_);
 574
 575     //get the number of indexes on this table
 576     //and populate the indexPtr array
 577     CatalogTableINDEX cIndex(systemDatabase_);
 578     table->numIndexes_ = cIndex.getNumIndexes(tptr);
 579     if (table->numIndexes_)
 580         table->indexPtr_ = new char*[table->numIndexes_];
 581     else
 582         table->indexPtr_ = NULL;
 583     cIndex.getIndexPtrs(tptr, table->indexPtr_);
 584
 585     //HACK:Below works only for one hash index on table
 586     if (table->numIndexes_ == 1)
 587     {
 588         SingleFieldHashIndexInfo *hIdxInfo = new SingleFieldHashIndexInfo();
 589         CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 590         cIndexField.getFieldNameAndType(table->indexPtr_[0], hIdxInfo->fldName,
 591                                                       hIdxInfo->type);
 592         ChunkIterator citer = CatalogTableINDEX::getIterator(table->indexPtr_[0]);
 593         hIdxInfo->noOfBuckets = CatalogTableINDEX::getNoOfBuckets(table->indexPtr_[0]);
 594         hIdxInfo->isUnique = CatalogTableINDEX::getUnique(table->indexPtr_[0]);
 595         hIdxInfo->buckets = (Bucket*)citer.nextElement();
 596
 597         table->idxInfo = (IndexInfo*) hIdxInfo;
 598     }
 599     systemDatabase_->releaseDatabaseMutex();
 600     if (tTuple->numFlds_ > 31)
 601     {
 602         table->isIntUsedForNULL = true;
 603         table->iNullInfo = 0;
 604     }
 605     else
 606     {
 607         table->isIntUsedForNULL = false;
 608         int noFields = os::align(tTuple->numFlds_);
 609         table->cNullInfo = (char*) malloc(noFields);
 610         for (int i =0 ; i < noFields; i++) table->cNullInfo[i] =0;
 611     }
 612     printDebug(DM_Database,"Opening table handle name:%s chunk:%x numIndex:%d",
 613                                          name, chunk, table->numIndexes_);
 614     logFinest(logger, "Opening Table %s" , name);
 615
 616     return table;
 617 }
 618
 619
 620
 621 List DatabaseManagerImpl::getAllTableNames()
 622 {
 623     DbRetVal ret = OK;
 624     //to store the tuple pointer of the table
 625     void *tptr =NULL;
 626
 627     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 628     if (OK != rv) {
 629         printError(ErrSysInternal, "Unable to get database mutex");
 630         List tableList;
 631         return tableList;
 632     }
 633     CatalogTableTABLE cTable(systemDatabase_);
 634     List tableList = cTable.getTableList();
 635     systemDatabase_->releaseDatabaseMutex();
 636     return tableList;
 637 }
 638
 639
 640
 641
 642 //Return values: -1 for table not found
 643 void DatabaseManagerImpl::closeTable(Table *table)
 644 {
 645     printDebug(DM_Database,"Closing table handle: %x", table);
 646     if (NULL == table) return;
 647     delete table;
 648     logFinest(logger, "Closing Table");
 649 }
 650
 651 DbRetVal DatabaseManagerImpl::createIndex(const char *indName, IndexInitInfo *info)
 652 {
 653     DbRetVal rv = OK;
 654     if (info->indType == hashIndex)
 655     {
 656         //Assumes info is of type HashIndexInitInfo
 657         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
 658         rv = createHashIndex(indName, info->tableName, info->list, hInfo->bucketSize,
 659                              info->isUnique);
 660     }
 661     else
 662     {
 663         //TODO::tree index
 664     }
 665     return rv;
 666 }
 667
 668
 669 //-1 -> Table does not exists
 670 //-2 -> Field does not exists
 671 //-3 -> bucketSize is not valid
 672 DbRetVal DatabaseManagerImpl::createHashIndex(const char *indName, const char *tblName,
 673                       FieldNameList &fldList, int bucketSize, bool isUnique)
 674 {
 675     //validate the bucket size
 676     if (bucketSize < 100 || bucketSize > 200000)
 677     {
 678         printError(ErrBadRange, "Index Bucket size %d not in range 100-200000",
 679                                  bucketSize);
 680         return ErrBadRange;
 681     }
 682     void *tptr =NULL;
 683     void *chunk = NULL;
 684     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 685     if (OK != rv)
 686     {
 687         printError(ErrSysInternal, "Unable to get database mutex");
 688         return ErrSysInternal;
 689     }
 690
 691     //check whether table exists
 692     CatalogTableTABLE cTable(systemDatabase_);
 693     cTable.getChunkAndTblPtr(tblName, chunk, tptr);
 694     if (NULL == tptr)
 695     {
 696         systemDatabase_->releaseDatabaseMutex();
 697         printError(ErrNotExists, "Table does not exist %s", tblName);
 698         return ErrNotExists;
 699     }
 700     int totFlds = fldList.size();
 701
 702     //check whether field exists
 703     char **fptr = new char* [totFlds];
 704     CatalogTableFIELD cField(systemDatabase_);
 705     rv = cField.getFieldPtrs(fldList, tptr, fptr);
 706     if (OK != rv)
 707     {
 708         delete[] fptr;
 709         systemDatabase_->releaseDatabaseMutex();
 710         printError(ErrNotExists, "Field does not exist");
 711         return ErrNotExists;
 712     }
 713
 714     //create chunk to store the meta data of the index created
 715     //for latches and bucket pointers
 716     Chunk* chunkInfo = createUserChunk(bucketSize * sizeof(Bucket));
 717     if (NULL == chunkInfo)
 718     {
 719         delete[] fptr;
 720         systemDatabase_->releaseDatabaseMutex();
 721         printError(ErrSysInternal, "Unable to create chunk");
 722         return ErrSysInternal;
 723     }
 724     //create memory for holding the bucket pointers
 725     void *buckets = chunkInfo->allocate(db_);
 726     if (NULL == buckets)
 727     {
 728         delete[] fptr;
 729         deleteUserChunk(chunkInfo);
 730         systemDatabase_->releaseDatabaseMutex();
 731         printError(ErrNoMemory, "Unable to allocate memory for bucket");
 732         return ErrNoMemory;
 733     }
 734     Bucket *buck = (Bucket*) buckets;
 735     initHashBuckets(buck, bucketSize);
 736
 737     //create chunk to store the hash index nodes
 738     Chunk* hChunk = createUserChunk(sizeof(HashIndexNode));
 739     if (NULL == hChunk)
 740     {
 741         delete[] fptr;
 742         deleteUserChunk(chunkInfo);
 743         systemDatabase_->releaseDatabaseMutex();
 744         printError(ErrSysInternal, "Unable to create chunk for storing hash index nodes");
 745         return ErrSysInternal;
 746     }
 747
 748
 749     //add row to INDEX
 750     void *tupleptr = NULL;
 751     CatalogTableINDEX cIndex(systemDatabase_);
 752     DbRetVal ret = cIndex.insert(indName, tptr, fldList.size(), isUnique,
 753                         chunkInfo, bucketSize, hChunk, tupleptr);
 754     if (OK != rv)
 755     {
 756         delete[] fptr;
 757         deleteUserChunk(hChunk);
 758         deleteUserChunk(chunkInfo);
 759         systemDatabase_->releaseDatabaseMutex();
 760         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
 761         return ErrSysInternal;
 762     }
 763     //add rows to INDEXFIELD
 764     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 765     ret = cIndexField.insert(fldList, tupleptr, tptr, fptr);
 766
 767     if (OK != rv)
 768     {
 769         delete[] fptr;
 770         deleteUserChunk(hChunk);
 771         deleteUserChunk(chunkInfo);
 772         systemDatabase_->releaseDatabaseMutex();
 773         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
 774         return ErrSysInternal;
 775     }
 776     delete[] fptr;
 777     //TODO::If tuples present in this table, then
 778     //create hash index nodes and store it
 779     //Take table lock
 780     systemDatabase_->releaseDatabaseMutex();
 781     printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x",
 782                                    indName, tblName, buckets);
 783     logFinest(logger, "Creating HashIndex %s on %s with bucket size %d",
 784                                    indName, tblName, buckets);
 785     return OK;
 786 }
 787
 788 void DatabaseManagerImpl::initHashBuckets(Bucket *buck, int bucketSize)
 789 {
 790     os::memset((void*)buck, 0, bucketSize * sizeof(Bucket));
 791
 792     for (int i=0; i < bucketSize ; i++)
 793     {
 794         buck[i].mutex_.init("Bucket");
 795     }
 796     return;
 797 }
 798
 799 DbRetVal DatabaseManagerImpl::dropIndex(const char *name)
 800 {
 801     return dropIndexInt(name, true);
 802 }
 803
 804 DbRetVal DatabaseManagerImpl::dropIndexInt(const char *name, bool takeLock)
 805 {
 806     DbRetVal rv = OK;
 807     void *chunk = NULL, *hchunk = NULL;
 808     void *tptr =NULL;
 809     int ret = 0;
 810     if (takeLock) {
 811         rv = systemDatabase_->getDatabaseMutex();
 812         if (OK != rv)
 813         {
 814             printError(ErrSysInternal, "Unable to get database mutex");
 815             return ErrSysInternal;
 816         }
 817     }
 818
 819     //remove the entry in INDEX
 820     CatalogTableINDEX cIndex(systemDatabase_);
 821     rv = cIndex.remove(name, chunk, hchunk, tptr);
 822     if (OK != rv)
 823     {
 824         if (takeLock) systemDatabase_->releaseDatabaseMutex();
 825         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
 826         return ErrSysInternal;
 827     }
 828     printDebug(DM_Database, "Removing from INDEX %s",name);
 829     //remove the entries in the INDEXFIELD table
 830     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 831     rv = cIndexField.remove(tptr);
 832     if (OK != rv)
 833     {
 834         if (takeLock) systemDatabase_->releaseDatabaseMutex();
 835         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
 836         return ErrSysInternal;
 837     }
 838     printDebug(DM_Database, "Removing from INDEXFIELD %s",name);
 839
 840     //delete the index chunk
 841     rv = deleteUserChunk((Chunk*)chunk);
 842     if (OK != rv)
 843     {
 844         if (takeLock) systemDatabase_->releaseDatabaseMutex();
 845         printError(ErrSysInternal, "Unable to delete the index chunk");
 846         return ErrSysInternal;
 847     }
 848     //delete the index hash node chunk
 849     rv = deleteUserChunk((Chunk*)hchunk);
 850     if (OK != rv)
 851     {
 852         if (takeLock) systemDatabase_->releaseDatabaseMutex();
 853         printError(ErrSysInternal, "Unable to delete the index hash node chunk");
 854         return ErrSysInternal;
 855     }
 856     if (takeLock) systemDatabase_->releaseDatabaseMutex();
 857     Chunk *chunkNode = systemDatabase_->getSystemDatabaseChunk(UserChunkTableId);
 858     chunkNode->free(systemDatabase_, (Chunk *) chunk);
 859     chunkNode->free(systemDatabase_, (Chunk *) hchunk);
 860
 861     //TODO::If tuples present in this table, then
 862     //free all hash index nodes for this table.
 863     //free all nodes in list of all buckets
 864     //Take table lock
 865
 866     printDebug(DM_Database, "Dropped hash index %s",name);
 867     logFinest(logger, "Deleted Index %s", name);
 868     return OK;
 869 }
 870
 871 DbRetVal DatabaseManagerImpl::registerThread()
 872 {
 873     DbRetVal rv = OK;
 874     if (pMgr_ != NULL)
 875     {
 876         printError(ErrAlready, "Process already registered\n");
 877         return ErrAlready;
 878     }
 879     pMgr_ = new ProcessManager();
 880     return pMgr_->registerThread();
 881 }
 882
 883 DbRetVal DatabaseManagerImpl::deregisterThread()
 884 {
 885     DbRetVal rv = OK;
 886     if (pMgr_ != NULL)
 887     {
 888         rv = pMgr_->deregisterThread();
 889         delete pMgr_;
 890         pMgr_ = NULL;
 891     }
 892     return rv;
 893 }
 894
 895 ChunkIterator DatabaseManagerImpl::getSystemTableIterator(CatalogTableID id)
 896 {
 897     Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(id);
 898     return fChunk->getIterator();
 899 }
 900
 901 Chunk* DatabaseManagerImpl::getSystemTableChunk(CatalogTableID id)
 902 {
 903     return systemDatabase_->getSystemDatabaseChunk(id);
 904 }