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