src/storage/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     Table *tbl = NULL;
  35     ListIterator iter = tableHandleList.getIterator();
  36     //PRABA::commented below...gives core dump
  37     //while ((tbl = (Table *)iter.nextElement()) != NULL) delete tbl;
  38     tableHandleList.reset();
  39     delete tMgr_;
  40     delete lMgr_;
  41 }
  42
  43 void DatabaseManagerImpl::createLockManager()
  44 {
  45     lMgr_ = new LockManager(systemDatabase_);
  46     return;
  47 }
  48
  49 void DatabaseManagerImpl::createTransactionManager()
  50 {
  51
  52     tMgr_ = new TransactionManager();
  53     tMgr_->setFirstTrans(systemDatabase_->getSystemDatabaseTrans(0));
  54     return;
  55 }
  56 void DatabaseManagerImpl::setProcSlot()
  57 {
  58 systemDatabase_->setProcSlot(procSlot);
  59 db_->setProcSlot(procSlot);
  60 }
  61 DbRetVal DatabaseManagerImpl::openSystemDatabase()
  62 {
  63     DbRetVal rv = openDatabase(SYSTEMDB);
  64     if (rv != OK) return rv;
  65     systemDatabase_ = db_;
  66     db_ = NULL;
  67     printDebug(DM_Database, "Opened system database");
  68     logFinest(logger, "Opened system database");
  69     return OK;
  70 }
  71
  72 DbRetVal DatabaseManagerImpl::closeSystemDatabase()
  73 {
  74     Database *db  = db_;
  75     //make them to point to system database file descriptor
  76     //and database pointer
  77     db_ = systemDatabase_;
  78     closeDatabase();
  79     db_ = db;
  80     printDebug(DM_Database, "Closed system database");
  81     logFinest(logger, "Closed System database");
  82     return OK;
  83 }
  84
  85 DbRetVal DatabaseManagerImpl::createDatabase(const char *name, size_t size)
  86 {
  87     if (NULL != db_ )
  88     {
  89         printError(ErrAlready, "Database is already created");
  90         return ErrAlready;
  91     }
  92     caddr_t rtnAddr = (caddr_t) NULL;
  93     shared_memory_id shm_id = 0;
  94
  95     char *startaddr = (char*)Conf::config.getMapAddress();
  96     shared_memory_key key = 0;
  97     if (0 == strcmp(name, SYSTEMDB))
  98     {
  99
 100         key = Conf::config.getSysDbKey();
 101     }
 102     else
 103     {
 104         startaddr = startaddr + Conf::config.getMaxSysDbSize();
 105         key = Conf::config.getUserDbKey();
 106     }
 107     shm_id = os::shm_create(key, size, 0666);
 108     if (-1 == shm_id)
 109     {
 110                 if (errno == EEXIST) {
 111                         printError(ErrOS, "Shared Memory already exists");
 112                 }
 113         printError(ErrOS, "Shared memory create failed");
 114         return ErrOS;
 115     }
 116
 117     void *shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 118     rtnAddr  = (caddr_t) shm_ptr;
 119     if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
 120     {
 121         printError(ErrOS, "Shared memory attach returned -ve value %d", rtnAddr);
 122         return ErrOS;
 123     }
 124     memset(shm_ptr, 0, size );
 125     db_ = new Database();
 126     printDebug(DM_Database, "Creating database:%s",name);
 127
 128     //TODO:for user database do not have transtable and processtable mutex
 129     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 130     db_->setDatabaseID(1);
 131     db_->setName(name);
 132     db_->setMaxSize(size);
 133     db_->setNoOfChunks(0);
 134     db_->initAllocDatabaseMutex();
 135     db_->initTransTableMutex();
 136     db_->initDatabaseMutex();
 137     db_->initProcessTableMutex();
 138     db_->setUniqueChunkID(100);
 139     //compute the first page after book keeping information
 140     size_t offset = os::alignLong(sizeof (DatabaseMetaData));
 141     //Only for system db chunk array, trans array and proc array will be there
 142     if (0 == strcmp(name, SYSTEMDB))
 143     {
 144         offset = offset + os::alignLong( MAX_CHUNKS  * sizeof (Chunk));
 145         offset = offset + os::alignLong( Conf::config.getMaxProcs()   * sizeof(Transaction));
 146         offset = offset + os::alignLong( Conf::config.getMaxProcs() * sizeof(ThreadInfo));
 147     }
 148     int multiple = os::floor(offset / PAGE_SIZE);
 149     char *curPage = (((char*)rtnAddr) + ((multiple + 1) * PAGE_SIZE));
 150
 151     db_->setCurrentPage(curPage);
 152     db_->setFirstPage(curPage);
 153
 154     if (0 == strcmp(name, SYSTEMDB)) return OK;
 155
 156     /*Allocate new chunk to store hash index nodes
 157     Chunk *chunkInfo = createUserChunk(sizeof(HashIndexNode));
 158     if (NULL == chunkInfo)
 159     {
 160         printError(ErrSysInternal, "Failed to allocate hash index nodes chunk");
 161         return ErrSysInternal;
 162     }
 163     printDebug(DM_Database, "Creating Chunk for storing Hash index nodes %x",
 164                                                                   chunkInfo);
 165
 166     db_->setHashIndexChunk(chunkInfo);*/
 167     logFinest(logger, "Created database %s" , name);
 168
 169     return OK;
 170 }
 171
 172 DbRetVal DatabaseManagerImpl::deleteDatabase(const char *name)
 173 {
 174     shared_memory_id shm_id = 0;
 175     if (0 == strcmp(name, SYSTEMDB))
 176     {
 177         shm_id = os::shm_open(Conf::config.getSysDbKey(), 100, 0666);
 178         os::shmctl(shm_id, IPC_RMID);
 179                 delete systemDatabase_;
 180                 systemDatabase_ = NULL;
 181     } else {
 182         shm_id = os::shm_open(Conf::config.getUserDbKey(), 100, 0666);
 183         os::shmctl(shm_id, IPC_RMID);
 184                 delete db_;
 185                 db_ = NULL;
 186     }
 187     logFinest(logger, "Deleted database %s" , name);
 188     return OK;
 189 }
 190
 191 DbRetVal DatabaseManagerImpl::openDatabase(const char *name)
 192 {
 193     long size = Conf::config.getMaxSysDbSize();
 194     char *startaddr = (char*)Conf::config.getMapAddress();
 195     if (0 == strcmp(name , SYSTEMDB))
 196     {
 197         if (NULL !=systemDatabase_)
 198         {
 199             printError(ErrAlready, "System Database already open");
 200             return ErrAlready;
 201         }
 202     }
 203     else
 204     {
 205         if (NULL ==systemDatabase_)
 206         {
 207             printError(ErrNotOpen, "System Database not open");
 208             return ErrNotOpen;
 209         }
 210         size = Conf::config.getMaxDbSize();
 211         startaddr = startaddr +  Conf::config.getMaxSysDbSize();
 212     }
 213     if (NULL != db_)
 214     {
 215         printError(ErrAlready, "User Database already open");
 216         return ErrAlready;
 217     }
 218     //system db should be opened before user database files
 219     caddr_t rtnAddr = (caddr_t) NULL;
 220
 221     shared_memory_id shm_id = 0;
 222     shared_memory_key key = 0;
 223
 224     if (0 == strcmp(name, SYSTEMDB))
 225         key = Conf::config.getSysDbKey();
 226     else
 227        key = Conf::config.getUserDbKey();
 228
 229
 230     int ret = ProcessManager::mutex.getLock(-1, false);
 231     //If you are not getting lock ret !=0, it means somebody else is there.
 232     //he will close the database.
 233     if (ret != 0)
 234     {
 235         printError(ErrSysInternal, "Another thread calling open:Wait and then Retry\n");
 236         return ErrSysInternal;
 237     }
 238     void *shm_ptr = NULL;
 239     bool firstThread = false;
 240     //printf("PRABA::DEBUG:: opendb %d %s\n", ProcessManager::noThreads, name);
 241     if (ProcessManager::noThreads == 0 && 0 == strcmp(name, SYSTEMDB)
 242        || ProcessManager::noThreads == 1 && 0 != strcmp(name, SYSTEMDB) ) {
 243        shm_id = os::shm_open(key, size, 0666);
 244        if (shm_id == -1 )
 245        {
 246            printError(ErrOS, "Shared memory open failed");
 247            ProcessManager::mutex.releaseLock(-1, false);
 248            return ErrOS;
 249        }
 250        shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 251        if (0 == strcmp(name, SYSTEMDB))
 252        {
 253              firstThread = true;
 254              ProcessManager::sysAddr = (char*) shm_ptr;
 255        }
 256        else
 257        {
 258               ProcessManager::usrAddr = (char*) shm_ptr;
 259        }
 260     } else {
 261         if (0 == strcmp(name, SYSTEMDB))
 262               shm_ptr = ProcessManager::sysAddr;
 263         else
 264               shm_ptr = ProcessManager::usrAddr;
 265     }
 266     ProcessManager::mutex.releaseLock(-1, false);
 267
 268
 269     rtnAddr  = (caddr_t) shm_ptr;
 270
 271     if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
 272     {
 273         printError(ErrOS, "Shared memory attach returned -ve value %x %d", shm_ptr, errno);
 274         return ErrOS;
 275     }
 276     db_ = new Database();
 277     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 278
 279     if (firstThread) ProcessManager::systemDatabase = db_;
 280
 281     printDebug(DM_Database, "Opening database: %s", name);
 282     logFinest(logger, "Opened database %s" , name);
 283     return OK;
 284 }
 285
 286 DbRetVal DatabaseManagerImpl::closeDatabase()
 287 {
 288
 289     if (NULL == db_)
 290     {
 291         //Database is already closed
 292         return OK;
 293     }
 294     printDebug(DM_Database, "Closing database: %s",(char*)db_->getName());
 295     //check if this is the last thread to be deregistered
 296     int ret = ProcessManager::mutex.getLock(-1, false);
 297     //If you are not getting lock ret !=0, it means somebody else is there.
 298     //he will close the database.
 299     if (ret == 0) {
 300     //printf("PRABA::FOR DEBUG closedb %d %s\n", ProcessManager::noThreads, (char*)db_->getName());
 301     if (ProcessManager::noThreads == 0 && 0 == strcmp((char*)db_->getName(), SYSTEMDB)
 302        || ProcessManager::noThreads == 1 && 0 != strcmp((char*)db_->getName(),  SYSTEMDB) )  {
 303                 os::shm_detach((char*)db_->getMetaDataPtr());
 304             }
 305     }
 306     ProcessManager::mutex.releaseLock(-1, false);
 307     logFinest(logger, "Closed database");
 308     delete db_;
 309     db_ = NULL;
 310     return OK;
 311 }
 312 //Assumes that system database mutex is taken before calling this.
 313 Chunk* DatabaseManagerImpl::createUserChunk(size_t size)
 314 {
 315     //Allocate new node in system database to store
 316     Chunk *chunk = getSystemTableChunk(UserChunkTableId);
 317     DbRetVal rv = OK;
 318     void *ptr = chunk->allocate(systemDatabase_, &rv);
 319     if (NULL == ptr)
 320     {
 321          printError(rv, "Allocation failed for User chunk catalog table");
 322          return NULL;
 323     }
 324     Chunk *chunkInfo = (Chunk*)ptr;
 325     chunkInfo->initMutex();
 326     if (0 != size) chunkInfo->setSize(size);
 327     if (chunkInfo->allocSize_ > PAGE_SIZE)
 328         chunkInfo->curPage_ = db_->getFreePage(chunkInfo->allocSize_);
 329     else
 330         chunkInfo->curPage_ = db_->getFreePage();
 331     if ( NULL == chunkInfo->curPage_)
 332     {
 333         chunkInfo->destroyMutex();
 334         chunk->free(db_, ptr);
 335         printError(ErrNoMemory, "Database full: No space to allocate from database");
 336         return NULL;
 337     }
 338     PageInfo* firstPageInfo = ((PageInfo*)chunkInfo->curPage_);
 339     if (chunkInfo->allocSize_ > PAGE_SIZE)
 340     {
 341         int multiple = os::floor(chunkInfo->allocSize_ / PAGE_SIZE);
 342         int offset = ((multiple + 1) * PAGE_SIZE);
 343         firstPageInfo->setPageAsUsed(offset);
 344     }
 345     else
 346     {
 347         firstPageInfo->setPageAsUsed(chunkInfo->allocSize_);
 348         char *data = ((char*)firstPageInfo) + sizeof(PageInfo);
 349         *(int*)data =0;
 350     }
 351     if (0 == size)
 352     {
 353         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
 354         varInfo->isUsed_ = 0;
 355         varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
 356
 357     }
 358     chunkInfo->firstPage_ = chunkInfo->curPage_;
 359
 360     if (0 == size)
 361         chunkInfo->setAllocType(VariableSizeAllocator);
 362     else
 363         chunkInfo->setAllocType(FixedSizeAllocator);
 364
 365     //TODO::Generate chunkid::use tableid
 366     chunkInfo->setChunkID(db_->getUniqueIDForChunk());
 367     db_->incrementChunk();
 368     printDebug(DM_Database, "Creating new User chunk chunkID:%d size: %d firstPage:%x",
 369                                -1, chunkInfo->allocSize_, firstPageInfo);
 370
 371     return chunkInfo;
 372 }
 373
 374 //Assumes that system database mutex is taken before calling this.
 375 DbRetVal DatabaseManagerImpl::deleteUserChunk(Chunk *chunk)
 376 {
 377     //Go to the pages and set them to notUsed
 378     Page *page = chunk->firstPage_;
 379     PageInfo* pageInfo = ((PageInfo*)page);
 380     //Here...sure that atleast one page will be there even no tuples
 381     //are inserted.so not checking if pageInfo == NULL
 382     while( pageInfo->nextPage_ != NULL)
 383     {
 384         PageInfo *prev = pageInfo;
 385         pageInfo = (PageInfo*)(pageInfo->nextPage_);
 386         //sets pageInfo->isUsed_ = 0 and pageInfo->hasFreeSpace_ = 0
 387         //and initializes the page content to zero
 388         if(NULL == pageInfo->nextPageAfterMerge_)
 389             os::memset(prev, 0, PAGE_SIZE);
 390         else
 391         {
 392             int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 393             os::memset(prev, 0, size);
 394         }
 395         printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, prev);
 396     }
 397     //The above loop wont execute for the last page
 398     //and for the case where table has only one page
 399     if(NULL == pageInfo->nextPageAfterMerge_)
 400         os::memset(pageInfo, 0, PAGE_SIZE);
 401     else
 402     {
 403         int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 404         os::memset(pageInfo, 0, size);
 405     }
 406     printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, pageInfo);
 407     chunk->chunkID_ = -1;
 408     chunk->allocSize_ = 0;
 409     chunk->curPage_ = NULL;
 410     chunk->firstPage_ = NULL;
 411     chunk->destroyMutex();
 412     db_->decrementChunk();
 413     Chunk *userChunk = getSystemTableChunk(UserChunkTableId);
 414     userChunk->free(systemDatabase_,chunk);
 415     printDebug(DM_Database,"deleting user chunk:%x",chunk);
 416     return OK;
 417 }
 418
 419 //-1 -> Unable to create chunk. No memory
 420 //-2 -> Unable to update the catalog tables
 421 DbRetVal DatabaseManagerImpl::createTable(const char *name, TableDef &def)
 422 {
 423     DbRetVal rv = OK;
 424     int fldCount = def.getFieldCount();
 425     if(0==fldCount)
 426     {
 427         printError(ErrNotExists,"Table can't be created without Field");
 428         return ErrNotExists;
 429     }
 430     //If total field count is less than 32, then 1 integer is used to store all null
 431     //information, if it is more then 1 char is used to store null information
 432     //of each field
 433     //This is to done to reduce cpu cycles for small tables
 434     int addSize = 0;
 435     if (fldCount < 31) addSize = 4; else addSize = os::align(fldCount);
 436     size_t sizeofTuple = os::align(def.getTupleSize())+addSize;
 437
 438     rv = systemDatabase_->getDatabaseMutex();
 439     if (OK != rv ) {
 440         printError(rv, "Unable to get Database mutex");
 441         return rv;
 442     }
 443
 444     void *tptr =NULL;
 445     void *chunk = NULL;
 446
 447     //check whether table already exists
 448     CatalogTableTABLE cTable(systemDatabase_);
 449     cTable.getChunkAndTblPtr(name, chunk, tptr);
 450     if (NULL != tptr)
 451     {
 452         systemDatabase_->releaseDatabaseMutex();
 453         printError(ErrAlready, "Table %s already exists", name);
 454         return ErrAlready;
 455     }
 456
 457     //create a chunk to store the tuples
 458     Chunk *ptr = createUserChunk(sizeofTuple);
 459     if (NULL  == ptr)
 460     {
 461         systemDatabase_->releaseDatabaseMutex();
 462         printError(ErrNoResource, "Unable to create user chunk");
 463         return ErrNoResource;
 464     }
 465     printDebug(DM_Database,"Created UserChunk:%x", ptr);
 466     ptr->setChunkName(name);
 467     //add row to TABLE
 468     int tblID = ((Chunk*)ptr)->getChunkID();
 469     rv = cTable.insert(name, tblID, sizeofTuple,
 470                                    def.getFieldCount(), ptr, tptr);
 471     if (OK != rv)
 472     {
 473         deleteUserChunk(ptr);
 474         systemDatabase_->releaseDatabaseMutex();
 475         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 476         return ErrSysInternal;
 477     }
 478     printDebug(DM_Database,"Inserted into TABLE:%s",name);
 479     //add rows to FIELD
 480     FieldIterator iter = def.getFieldIterator();
 481     CatalogTableFIELD cField(systemDatabase_);
 482     rv = cField.insert(iter, tblID ,tptr);
 483     if (OK != rv)
 484     {
 485         deleteUserChunk(ptr);
 486         void *cptr, *ttptr;//Dummy as remove below needs both these OUT params
 487         cTable.remove(name, cptr, ttptr);
 488         systemDatabase_->releaseDatabaseMutex();
 489         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 490         return ErrSysInternal;
 491     }
 492     printDebug(DM_Database,"Inserted into FIELD:%s",name);
 493     systemDatabase_->releaseDatabaseMutex();
 494     printDebug(DM_Database,"Table Created:%s",name);
 495     logFinest(logger, "Table Created %s" , name);
 496     return OK;
 497 }
 498
 499 //TODO::If any operation fails in between, then we may have some
 500 //dangling tuples, say we have have rows in INDEX table
 501 //which will not have any corresponding entries in TABLE
 502 //CHANGE the sequence so that it deletes from the bottom as
 503 //opposed to start from top as is written now
 504 DbRetVal DatabaseManagerImpl::dropTable(const char *name)
 505 {
 506     void *chunk = NULL;
 507     void *tptr =NULL;
 508     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 509     if (OK != rv) {
 510         printError(ErrSysInternal, "Unable to get database mutex");
 511         return ErrSysInternal;
 512     }
 513
 514     //remove the entry in TABLE
 515     CatalogTableTABLE cTable(systemDatabase_);
 516     rv = cTable.getChunkAndTblPtr(name, chunk, tptr);
 517     if (OK != rv) {
 518         systemDatabase_->releaseDatabaseMutex();
 519         printError(ErrSysInternal, "Table %s does not exist", name);
 520         return ErrSysInternal;
 521     }
 522     rv = lMgr_->getExclusiveLock(chunk, NULL);
 523     if (rv !=OK)
 524     {
 525         systemDatabase_->releaseDatabaseMutex();
 526         printError(ErrLockTimeOut, "Unable to acquire exclusive lock on the table\n");
 527         return rv;
 528     }
 529
 530     rv = cTable.remove(name, chunk, tptr);
 531     if (OK != rv) {
 532         systemDatabase_->releaseDatabaseMutex();
 533         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 534         return ErrSysInternal;
 535     }
 536     printDebug(DM_Database,"Deleted from TABLE:%s",name);
 537
 538     //remove the entries in the FIELD table
 539     CatalogTableFIELD cField(systemDatabase_);
 540     rv = cField.remove(tptr);
 541     if (OK != rv) {
 542         systemDatabase_->releaseDatabaseMutex();
 543         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 544         return ErrSysInternal;
 545     }
 546     printDebug(DM_Database,"Deleted from FIELD:%s",name);
 547
 548     rv = deleteUserChunk((Chunk*)chunk);
 549     if (OK != rv) {
 550         systemDatabase_->releaseDatabaseMutex();
 551         printError(rv, "Unable to delete the chunk");
 552         return rv;
 553     }
 554     printDebug(DM_Database,"Deleted UserChunk:%x", chunk);
 555
 556     //TODO::check whether indexes are available and drop that also.
 557     CatalogTableINDEX cIndex(systemDatabase_);
 558     int noIndexes = cIndex.getNumIndexes(tptr);
 559     for (int i =1 ; i<= noIndexes; i++) {
 560         char *idxName = cIndex.getIndexName(tptr, 1);
 561         dropIndexInt(idxName, false);
 562     }
 563     Chunk *chunkNode = systemDatabase_->getSystemDatabaseChunk(UserChunkTableId);
 564     chunkNode->free(systemDatabase_, (Chunk *) chunk);
 565     systemDatabase_->releaseDatabaseMutex();
 566     printDebug(DM_Database, "Deleted Table %s" , name);
 567     logFinest(logger, "Deleted Table %s" , name);
 568     rv = lMgr_->releaseLock(chunk);
 569     if (rv !=OK)
 570     {
 571         printError(ErrLockTimeOut, "Unable to release exclusive lock on the table\n");
 572         return rv;
 573     }
 574     return OK;
 575 }
 576
 577 //Return values: NULL for table not found
 578 Table* DatabaseManagerImpl::openTable(const char *name)
 579 {
 580     DbRetVal ret = OK;
 581     //TODO::store table handles in list so that if it is
 582     //not closed by the application. destructor shall close it.
 583     TableImpl *table = new TableImpl();
 584     table->setDB(db_);
 585     table->setSystemDB(systemDatabase_);
 586     table->setLockManager(lMgr_);
 587     table->setTrans(ProcessManager::getThreadTransAddr(systemDatabase_->procSlot));
 588
 589     //to store the chunk pointer of table
 590     void *chunk = NULL;
 591
 592     //to store the tuple pointer of the table
 593     void *tptr =NULL;
 594
 595     //TODO::need to take shared lock on the table so that
 596     //all ddl operation will be denied on that table
 597     //which includes index creation, alter table
 598
 599     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 600     if (OK != rv) {
 601         printError(ErrSysInternal, "Unable to get database mutex");
 602         delete table;
 603         return NULL;
 604     }
 605     CatalogTableTABLE cTable(systemDatabase_);
 606     ret = cTable.getChunkAndTblPtr(name, chunk, tptr);
 607     if ( OK != ret)
 608     {
 609         systemDatabase_->releaseDatabaseMutex();
 610         delete table;
 611         printError(ErrNotExists, "Table not exists %s", name);
 612         return NULL;
 613     }
 614     CTABLE *tTuple = (CTABLE*)tptr;
 615     table->setTableInfo(tTuple->tblName_, tTuple->tblID_, tTuple->length_,
 616                         tTuple->numFlds_, tTuple->numIndexes_, tTuple->chunkPtr_);
 617     /*rv = table->lock(true); //take shared lock
 618     if (rv !=OK)
 619     {
 620         printError(ErrLockTimeOut, "Unable to acquire shared lock on the table\n");
 621         systemDatabase_->releaseDatabaseMutex();
 622         delete table;
 623         return NULL;
 624     }*/
 625
 626
 627     if (tTuple->numFlds_ < 31)
 628     {
 629         table->isIntUsedForNULL = true;
 630         table->iNullInfo = 0;
 631         table->iNotNullInfo =0;
 632     }
 633     else
 634     {
 635         table->isIntUsedForNULL = false;
 636         int noFields = os::align(tTuple->numFlds_);
 637         table->cNullInfo = (char*) malloc(noFields);
 638         table->cNotNullInfo = (char*) malloc(noFields);
 639         for (int i =0 ; i < noFields; i++) table->cNullInfo[i] =0;
 640         for (int i =0 ; i < noFields; i++) table->cNotNullInfo[i] =0;
 641
 642     }
 643
 644     //get field information from FIELD table
 645     CatalogTableFIELD cField(systemDatabase_);
 646     cField.getFieldInfo(tptr, table->fldList_);
 647
 648     //populate the notnull info
 649     FieldIterator fIter = table->fldList_.getIterator();
 650     int fldpos=1;
 651     while (fIter.hasElement())
 652     {
 653         FieldDef def = fIter.nextElement();
 654         if (table->isIntUsedForNULL) {
 655             if (def.isNull_) SETBIT(table->iNotNullInfo, fldpos);
 656         }
 657         else {
 658             if (def.isNull_) table->cNotNullInfo[fldpos-1] = 1;
 659         }
 660         fldpos++;
 661    }
 662
 663     //get the number of indexes on this table
 664     //and populate the indexPtr array
 665     CatalogTableINDEX cIndex(systemDatabase_);
 666     table->numIndexes_ = cIndex.getNumIndexes(tptr);
 667     if (table->numIndexes_) {
 668         table->indexPtr_ = new char*[table->numIndexes_];
 669         table->idxInfo = new IndexInfo*[table->numIndexes_];
 670     }
 671     else
 672     {
 673         table->indexPtr_ = NULL;
 674     }
 675     cIndex.getIndexPtrs(tptr, table->indexPtr_);
 676     for (int i =0 ; i < table->numIndexes_; i++ )
 677     {
 678         HashIndexInfo *hIdxInfo = new HashIndexInfo();
 679         CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 680         cIndexField.getFieldInfo(table->indexPtr_[i], hIdxInfo->idxFldList);
 681         ChunkIterator citer = CatalogTableINDEX::getIterator(table->indexPtr_[i]);
 682         hIdxInfo->indexPtr = table->indexPtr_[i];
 683         hIdxInfo->indType = ((CINDEX*)hIdxInfo->indexPtr)->indexType_;
 684         hIdxInfo->noOfBuckets = CatalogTableINDEX::getNoOfBuckets(table->indexPtr_[i]);
 685         FieldIterator fIter = hIdxInfo->idxFldList.getIterator();
 686         bool firstFld = true;
 687         while (fIter.hasElement())
 688         {
 689             FieldDef def = fIter.nextElement();
 690             if (firstFld)
 691             {
 692                 hIdxInfo->fldOffset = table->fldList_.getFieldOffset(def.fldName_);
 693                 hIdxInfo->type = table->fldList_.getFieldType(def.fldName_);
 694                 hIdxInfo->compLength = table->fldList_.getFieldLength(def.fldName_);
 695                 firstFld = false;
 696             }else {
 697                 hIdxInfo->type = typeComposite;
 698                 hIdxInfo->compLength = hIdxInfo->compLength +
 699                          table->fldList_.getFieldLength(def.fldName_);
 700             }
 701         }
 702
 703         hIdxInfo->isUnique = CatalogTableINDEX::getUnique(table->indexPtr_[i]);
 704         hIdxInfo->buckets = (Bucket*)citer.nextElement();
 705         table->idxInfo[i] = (IndexInfo*) hIdxInfo;
 706     }
 707     systemDatabase_->releaseDatabaseMutex();
 708    // lMgr->  tTuple->chunkPtr_
 709     printDebug(DM_Database,"Opening table handle name:%s chunk:%x numIndex:%d",
 710                                          name, chunk, table->numIndexes_);
 711     logFinest(logger, "Opening Table %s" , name);
 712
 713     tableHandleList.append(table);
 714
 715     return table;
 716 }
 717
 718
 719
 720 List DatabaseManagerImpl::getAllTableNames()
 721 {
 722     DbRetVal ret = OK;
 723     //to store the tuple pointer of the table
 724     void *tptr =NULL;
 725
 726     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 727     if (OK != rv) {
 728         printError(ErrSysInternal, "Unable to get database mutex");
 729         List tableList;
 730         return tableList;
 731     }
 732     CatalogTableTABLE cTable(systemDatabase_);
 733     List tableList = cTable.getTableList();
 734     systemDatabase_->releaseDatabaseMutex();
 735     return tableList;
 736 }
 737
 738
 739
 740
 741 //Return values: -1 for table not found
 742 void DatabaseManagerImpl::closeTable(Table *table)
 743 {
 744     printDebug(DM_Database,"Closing table handle: %x", table);
 745     if (NULL == table) return;
 746     //table->unlock();
 747     tableHandleList.remove(table, false);
 748     logFinest(logger, "Closing Table");
 749 }
 750
 751 DbRetVal DatabaseManagerImpl::createIndex(const char *indName, IndexInitInfo *info)
 752 {
 753     DbRetVal rv = OK;
 754     if (!info->isUnique && info->isPrimary)
 755     {
 756         printError(ErrBadCall, "Primary key cannot be non unique\n");
 757         return ErrBadCall;
 758     }
 759     if (info->indType == hashIndex)
 760     {
 761         //Assumes info is of type HashIndexInitInfo
 762         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
 763         rv = createHashIndex(indName, info->tableName, info->list, hInfo->bucketSize,
 764                              info->isUnique, info->isPrimary);
 765     }
 766     else if (info->indType == treeIndex)
 767     {
 768         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
 769         rv = createTreeIndex(indName, info->tableName, info->list,
 770                           hInfo->bucketSize, info->isUnique, info->isPrimary);
 771
 772     }else {
 773         printError(ErrBadCall, "Index type not supported\n");
 774         return ErrBadCall;
 775     }
 776     return rv;
 777 }
 778
 779
 780 //-1 -> Table does not exists
 781 //-2 -> Field does not exists
 782 //-3 -> bucketSize is not valid
 783 DbRetVal DatabaseManagerImpl::createHashIndex(const char *indName, const char *tblName,
 784                       FieldNameList &fldList, int bucketSize, bool isUnique, bool isPrimary)
 785 {
 786     //validate the bucket size
 787     if (bucketSize < 100 || bucketSize > 200000)
 788     {
 789         printError(ErrBadRange, "Index Bucket size %d not in range 100-200000",
 790                                  bucketSize);
 791         return ErrBadRange;
 792     }
 793     int totFlds = fldList.size();
 794     if (totFlds == 0)
 795     {
 796         printError(ErrBadCall, "No Field name specified");
 797         return ErrBadCall;
 798     }
 799     void *tptr =NULL;
 800     void *chunk = NULL;
 801     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 802     if (OK != rv)
 803     {
 804         printError(ErrSysInternal, "Unable to get database mutex");
 805         return ErrSysInternal;
 806     }
 807
 808     //check whether table exists
 809     CatalogTableTABLE cTable(systemDatabase_);
 810     cTable.getChunkAndTblPtr(tblName, chunk, tptr);
 811     if (NULL == tptr)
 812     {
 813         systemDatabase_->releaseDatabaseMutex();
 814         printError(ErrNotExists, "Table does not exist %s", tblName);
 815         return ErrNotExists;
 816     }
 817
 818     //check whether field exists
 819     char **fptr = new char* [totFlds];
 820     CatalogTableFIELD cField(systemDatabase_);
 821     rv = cField.getFieldPtrs(fldList, tptr, fptr);
 822     if (OK != rv)
 823     {
 824         delete[] fptr;
 825         systemDatabase_->releaseDatabaseMutex();
 826         //TODO::check test cases of dbapi/Index, they give wrong results
 827         //if (rv == ErrBadCall) {
 828         ////    if (isPrimary) printError(ErrBadCall, "Field can have NULL values");
 829         //} else {
 830         //printError(ErrNotExists, "Field does not exist");
 831         //}
 832         //return ErrBadCall;
 833         if (rv != ErrBadCall) {
 834             printError(ErrNotExists, "Field does not exist");
 835             return ErrNotExists;
 836         }
 837     }
 838     for (int i=0; i <totFlds; i++)
 839     {
 840         CFIELD* fInfo = (CFIELD*)fptr[i];
 841         if (fInfo->type_ == typeFloat || fInfo->type_ == typeDouble || fInfo->type_ == typeTimeStamp)
 842         {
 843             printError(ErrBadArg, "HashIndex cannot be created for float or double or timestamp type");
 844             delete[] fptr;
 845             systemDatabase_->releaseDatabaseMutex();
 846             return ErrBadArg;
 847         }
 848         if (!fInfo->isNull_ && isPrimary )
 849         {
 850             printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
 851             delete[] fptr;
 852             systemDatabase_->releaseDatabaseMutex();
 853             return ErrBadArg;
 854         }
 855          if(isPrimary){fInfo->isPrimary_=true;fInfo->isUnique_=true;}
 856          if(isUnique){fInfo->isUnique_=true;}
 857     }
 858     //create chunk to store the meta data of the index created
 859     //for latches and bucket pointers
 860     printDebug(DM_HashIndex, "Creating chunk for storing hash buckets of size %d\n",
 861                              bucketSize * sizeof(Bucket));
 862     Chunk* chunkInfo = createUserChunk(bucketSize * sizeof(Bucket));
 863     if (NULL == chunkInfo)
 864     {
 865         delete[] fptr;
 866         systemDatabase_->releaseDatabaseMutex();
 867         printError(ErrSysInternal, "Unable to create chunk");
 868         return ErrSysInternal;
 869     }
 870     chunkInfo->setChunkName(indName);
 871     //create memory for holding the bucket pointers
 872     void *buckets = chunkInfo->allocate(db_, &rv);
 873     if (NULL == buckets)
 874     {
 875         delete[] fptr;
 876         deleteUserChunk(chunkInfo);
 877         systemDatabase_->releaseDatabaseMutex();
 878         printError(rv, "Unable to allocate memory for bucket");
 879         return rv;
 880     }
 881     Bucket *buck = (Bucket*) buckets;
 882     initHashBuckets(buck, bucketSize);
 883
 884     //create chunk to store the hash index nodes
 885     Chunk* hChunk = createUserChunk(sizeof(HashIndexNode));
 886     if (NULL == hChunk)
 887     {
 888         delete[] fptr;
 889         deleteUserChunk(chunkInfo);
 890         systemDatabase_->releaseDatabaseMutex();
 891         printError(ErrSysInternal, "Unable to create chunk for storing hash index nodes");
 892         return ErrSysInternal;
 893     }
 894      hChunk->setChunkName(indName);
 895     //add row to INDEX
 896     void *tupleptr = NULL;
 897     CatalogTableINDEX cIndex(systemDatabase_);
 898     rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
 899                         chunkInfo, bucketSize, hChunk, tupleptr);
 900     if (OK != rv)
 901     {
 902         delete[] fptr;
 903         deleteUserChunk(hChunk);
 904         deleteUserChunk(chunkInfo);
 905         systemDatabase_->releaseDatabaseMutex();
 906         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
 907         return ErrSysInternal;
 908     }
 909     //add rows to INDEXFIELD
 910     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 911     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
 912
 913     if (OK != rv)
 914     {
 915         delete[] fptr;
 916         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
 917         deleteUserChunk(hChunk);
 918         deleteUserChunk(chunkInfo);
 919         systemDatabase_->releaseDatabaseMutex();
 920         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
 921         return ErrSysInternal;
 922     }
 923     delete[] fptr;
 924     systemDatabase_->releaseDatabaseMutex();
 925
 926     //TODO:: Take table lock
 927
 928     // Following code is written by Kishor Amballi
 929     TableImpl *tbl = (TableImpl *) openTable(tblName);
 930     if (! tbl->numTuples()) {
 931         printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x", indName, tblName, buckets);
 932         logFinest(logger, "Creating HashIndex %s on %s with bucket size %d", indName, tblName, buckets);
 933         closeTable(tbl);
 934         return OK;
 935     }
 936     HashIndexInfo *indxInfo = NULL;
 937     int i = 0;
 938     for (i = 0; i < tbl->numIndexes_; i++) {
 939         if(((HashIndexInfo *)tbl->idxInfo[i])->indexPtr == tupleptr) {
 940             indxInfo = (HashIndexInfo *) tbl->idxInfo[i];
 941             break;
 942         }
 943     }
 944     void *recPtr = NULL;
 945     ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
 946     while ((recPtr = chIter.nextElement()) != NULL) {
 947         rv = tbl->insertIndexNode(*tbl->trans, tupleptr, indxInfo, recPtr);
 948         if (rv == ErrUnique) {
 949             closeTable(tbl);
 950             dropIndex(indName);
 951             return rv;
 952         }
 953     }
 954     closeTable(tbl);
 955     printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x", indName, tblName, buckets);
 956     logFinest(logger, "Creating HashIndex %s on %s with bucket size %d", indName, tblName, buckets);
 957     return OK;
 958 }
 959
 960
 961 DbRetVal DatabaseManagerImpl::createTreeIndex(const char *indName, const char *tblName,
 962                       FieldNameList &fldList, int nodeSize, bool isUnique, bool isPrimary)
 963 {
 964     if (nodeSize < 20 || nodeSize > 20000)
 965     {
 966         printError(ErrBadRange,"Tree Index Node size %d not in range 20-20000",
 967                                  nodeSize);
 968         return ErrBadRange;
 969     }
 970     int totFlds = fldList.size();
 971     if (totFlds == 0)
 972     {
 973         printError(ErrBadCall, "No Field name specified");
 974         return ErrBadCall;
 975     }
 976     void *tptr =NULL;
 977     void *chunk = NULL;
 978     DbRetVal rv = systemDatabase_->getDatabaseMutex();
 979     if (OK != rv)
 980     {
 981         printError(ErrSysInternal, "Unable to get database mutex");
 982         return ErrSysInternal;
 983     }
 984     //check whether table exists
 985
 986     CatalogTableTABLE cTable(systemDatabase_);
 987     cTable.getChunkAndTblPtr(tblName, chunk, tptr);
 988     if (NULL == tptr)
 989     {
 990         systemDatabase_->releaseDatabaseMutex();
 991         printError(ErrNotExists, "Table does not exist %s", tblName);
 992         return ErrNotExists;
 993     }
 994     char **fptr = new char* [totFlds];
 995     CatalogTableFIELD cField(systemDatabase_);
 996     rv = cField.getFieldPtrs(fldList, tptr, fptr);
 997     if (OK != rv)
 998     {
 999         delete[] fptr;
1000         systemDatabase_->releaseDatabaseMutex();
1001         if (rv != ErrBadCall) {
1002             printError(ErrNotExists, "Field does not exist");
1003             return ErrNotExists;
1004         }
1005     }
1006     for (int i=0; i <totFlds; i++)
1007     {
1008         CFIELD* fInfo = (CFIELD*)fptr[i];
1009         if (!fInfo->isNull_ && isPrimary )
1010         {
1011             printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
1012             delete[] fptr;
1013             systemDatabase_->releaseDatabaseMutex();
1014             return ErrBadArg;
1015         }
1016     }
1017     int chunkSize = sizeof(TreeNode)+(nodeSize * sizeof(void*));
1018     printDebug(DM_HashIndex, "Creating chunk for storing tree nodes of size %d\n", chunkSize);
1019
1020     Chunk* chunkInfo = createUserChunk(chunkSize);
1021     if (NULL == chunkInfo)
1022     {
1023         delete[] fptr;
1024         systemDatabase_->releaseDatabaseMutex();
1025         printError(ErrSysInternal, "Unable to create chunk");
1026         return ErrSysInternal;
1027     }
1028
1029
1030     void *tupleptr = NULL;
1031
1032     CatalogTableINDEX cIndex(systemDatabase_);
1033     rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
1034                         chunkInfo, nodeSize, NULL, tupleptr);
1035     if (OK != rv)
1036     {
1037         delete[] fptr;
1038         deleteUserChunk(chunkInfo);
1039         systemDatabase_->releaseDatabaseMutex();
1040         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
1041         return ErrSysInternal;
1042     }
1043     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
1044     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
1045
1046     if (OK != rv)
1047     {
1048         delete[] fptr;
1049         void *hChunk = NULL;
1050         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
1051         deleteUserChunk(chunkInfo);
1052         systemDatabase_->releaseDatabaseMutex();
1053         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
1054         return ErrSysInternal;
1055     }
1056     delete[] fptr;
1057     systemDatabase_->releaseDatabaseMutex();
1058     //TODO::if tuples already present in this table, then create tree index '
1059     //nodes
1060     TableImpl *tbl = (TableImpl *) openTable(tblName);
1061     if (! tbl->numTuples()) {
1062         printDebug(DM_Database, "Creating Tree Index Name:%s tblname:%s node size:%x",indName, tblName, nodeSize);
1063         logFinest(logger, "Creating TreeIndex %s on %s with node size %d",indName, tblName, nodeSize);
1064         return OK;
1065     }
1066     HashIndexInfo *indxInfo = NULL;
1067     int i = 0;
1068     for (i = 0; i < tbl->numIndexes_; i++) {
1069         if(((HashIndexInfo *)tbl->idxInfo[i])->indexPtr == tupleptr) {
1070             indxInfo = (HashIndexInfo *) tbl->idxInfo[i];
1071             break;
1072         }
1073     }
1074     void *recPtr = NULL;
1075     ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
1076     while ((recPtr = chIter.nextElement()) != NULL) {
1077         rv = tbl->insertIndexNode(*tbl->trans, tupleptr, indxInfo, recPtr);
1078         if (rv == ErrUnique) {
1079             closeTable(tbl);
1080             dropIndex(indName);
1081             closeTable(tbl);
1082             return rv;
1083         }
1084     }
1085     closeTable(tbl);
1086     printDebug(DM_Database, "Creating Tree Index Name:%s tblname:%s node size:%x",
1087                                    indName, tblName, nodeSize);
1088     logFinest(logger, "Creating TreeIndex %s on %s with node size %d",
1089                                    indName, tblName, nodeSize);
1090     return OK;
1091 }
1092
1093
1094
1095 void DatabaseManagerImpl::initHashBuckets(Bucket *buck, int bucketSize)
1096 {
1097     os::memset((void*)buck, 0, bucketSize * sizeof(Bucket));
1098
1099     for (int i=0; i < bucketSize ; i++)
1100     {
1101         buck[i].mutex_.init("Bucket");
1102     }
1103     return;
1104 }
1105
1106 DbRetVal DatabaseManagerImpl::dropIndex(const char *name)
1107 {
1108     return dropIndexInt(name, true);
1109 }
1110
1111 DbRetVal DatabaseManagerImpl::dropIndexInt(const char *name, bool takeLock)
1112 {
1113     DbRetVal rv = OK;
1114     void *chunk = NULL, *hchunk = NULL;
1115     void *tptr =NULL;
1116     int ret = 0;
1117     if (takeLock) {
1118         rv = systemDatabase_->getDatabaseMutex();
1119         if (OK != rv)
1120         {
1121             printError(ErrSysInternal, "Unable to get database mutex");
1122             return ErrSysInternal;
1123         }
1124     }
1125
1126     //remove the entry in INDEX
1127     CatalogTableINDEX cIndex(systemDatabase_);
1128     rv = cIndex.remove(name, chunk, hchunk, tptr);
1129     if (OK != rv)
1130     {
1131         if (takeLock) systemDatabase_->releaseDatabaseMutex();
1132         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
1133         return ErrSysInternal;
1134     }
1135     printDebug(DM_Database, "Removing from INDEX %s",name);
1136     //remove the entries in the INDEXFIELD table
1137     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
1138     rv = cIndexField.remove(tptr);
1139     if (OK != rv)
1140     {
1141         if (takeLock) systemDatabase_->releaseDatabaseMutex();
1142         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
1143         return ErrSysInternal;
1144     }
1145     printDebug(DM_Database, "Removing from INDEXFIELD %s",name);
1146
1147     //delete the index chunk
1148     CINDEX *iptr = (CINDEX*)tptr;
1149     rv = deleteUserChunk((Chunk*)chunk);
1150     if (OK != rv)
1151     {
1152         if (takeLock) systemDatabase_->releaseDatabaseMutex();
1153         printError(ErrSysInternal, "Unable to delete the index chunk");
1154         return ErrSysInternal;
1155     }
1156     //delete the index hash node chunk
1157     if (iptr->indexType_ == hashIndex) {
1158         rv = deleteUserChunk((Chunk*)hchunk);
1159         if (OK != rv)
1160         {
1161             if (takeLock) systemDatabase_->releaseDatabaseMutex();
1162             printError(ErrSysInternal, "Unable to delete the index hash node chunk");
1163             return ErrSysInternal;
1164         }
1165     }
1166     if (takeLock) systemDatabase_->releaseDatabaseMutex();
1167     Chunk *chunkNode = systemDatabase_->getSystemDatabaseChunk(UserChunkTableId);
1168     chunkNode->free(systemDatabase_, (Chunk *) chunk);
1169     if (iptr->indexType_ == hashIndex) {
1170         chunkNode->free(systemDatabase_, (Chunk *) hchunk);
1171     }
1172
1173     //TODO::If tuples present in this table, then
1174     //free all hash index nodes for this table.
1175     //free all nodes in list of all buckets
1176     //Take table lock
1177
1178     printDebug(DM_Database, "Dropped hash index %s",name);
1179     logFinest(logger, "Deleted Index %s", name);
1180     return OK;
1181 }
1182 DbRetVal DatabaseManagerImpl::printIndexInfo(char *name)
1183 {
1184     CatalogTableINDEX cIndex(systemDatabase_);
1185     DbRetVal rv = OK;
1186     void *chunk = NULL, *hchunk = NULL;
1187     void *tptr =NULL;
1188     rv = cIndex.get(name, chunk, hchunk, tptr);
1189     if (OK != rv) return rv;
1190     printf("<IndexName> %s </IndexName>\n", name);
1191     printf("<Unique> %d </Unique>\n", CatalogTableINDEX::getUnique(tptr));
1192     Chunk *ch = (Chunk*) chunk;
1193     printf("<HashBucket>\n");
1194     printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
1195     printf("  <TotalBuckets> %d </TotalBuckets> \n", CatalogTableINDEX::getNoOfBuckets(tptr));
1196     printf("</HashBucket>\n");
1197
1198     ch = (Chunk*) hchunk;
1199     printf("<IndexNodes>\n");
1200     printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
1201     printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
1202     printf("<IndexNodes>\n");
1203     return OK;
1204 }
1205
1206 DbRetVal DatabaseManagerImpl::registerThread()
1207 {
1208     DbRetVal rv = OK;
1209     if (pMgr_ != NULL)
1210     {
1211         printError(ErrAlready, "Process already registered\n");
1212         return ErrAlready;
1213     }
1214     pMgr_ = new ProcessManager();
1215     rv = pMgr_->registerThread();
1216     if (rv ==OK) { procSlot = pMgr_->getProcSlot();
1217         printDebug(DM_Process, "Process registed with slot %d\n", procSlot);
1218     }
1219     return rv;
1220 }
1221
1222 DbRetVal DatabaseManagerImpl::deregisterThread()
1223 {
1224     DbRetVal rv = OK;
1225     if (pMgr_ != NULL)
1226     {
1227         rv = pMgr_->deregisterThread(procSlot);
1228         delete pMgr_;
1229         pMgr_ = NULL;
1230     }
1231     return rv;
1232 }
1233
1234 bool DatabaseManagerImpl::isAnyOneRegistered()
1235 {
1236     if (pMgr_ != NULL) return pMgr_->isAnyOneRegistered();
1237     return true;
1238 }
1239
1240
1241 void DatabaseManagerImpl::printUsageStatistics()
1242 {
1243     pMgr_->printUsageStatistics();
1244     tMgr_->printUsageStatistics();
1245     lMgr_->printUsageStatistics();
1246 }
1247
1248 void DatabaseManagerImpl::printDebugLockInfo()
1249 {
1250     lMgr_->printDebugInfo();
1251 }
1252
1253 void DatabaseManagerImpl::printDebugTransInfo()
1254 {
1255     tMgr_->printDebugInfo(systemDatabase_);
1256 }
1257 void DatabaseManagerImpl::printDebugProcInfo()
1258 {
1259     pMgr_->printDebugInfo();
1260 }
1261 void DatabaseManagerImpl::printDebugChunkInfo()
1262 {
1263     printf("<NotYetImplemented> </NotYetImplemented>\n");
1264 }
1265 ChunkIterator DatabaseManagerImpl::getSystemTableIterator(CatalogTableID id)
1266 {
1267     Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(id);
1268     return fChunk->getIterator();
1269 }
1270
1271 Chunk* DatabaseManagerImpl::getSystemTableChunk(CatalogTableID id)
1272 {
1273     return systemDatabase_->getSystemDatabaseChunk(id);
1274 }