src/server/DatabaseManagerImpl.cxx

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