src/storage/Database.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<os.h>
  18 #include<CatalogTables.h>
  19 #include<Transaction.h>
  20 #include<Lock.h>
  21 #include<Debug.h>
  22 #include<Config.h>
  23 #include<Process.h>
  24 #include<HeapAllocator.h>
  25
  26 const char* Database::getName()
  27 {
  28     return metaData_->dbName_;
  29 }
  30
  31 int Database::getDatabaseID()
  32 {
  33     return metaData_->dbID_;
  34 }
  35
  36 long Database::getMaxSize()
  37 {
  38     return metaData_->maxSize_;
  39 }
  40
  41 long Database::getCurrentSize()
  42 {
  43     return metaData_->curSize_;
  44
  45 }
  46
  47 Page* Database::getCurrentPage()
  48 {
  49     return metaData_->curPage_;
  50
  51 }
  52
  53 Page* Database::getFirstPage()
  54 {
  55     return metaData_->firstPage_;
  56
  57 }
  58
  59 int Database::getNoOfChunks()
  60 {
  61     return metaData_->noOfChunks_;
  62 }
  63
  64 Chunk* Database::getHashIndexChunk()
  65 {
  66     return metaData_->hashIndexChunk_;
  67 }
  68
  69 void Database::setDatabaseID(int id)
  70 {
  71     metaData_->dbID_ = id;
  72 }
  73
  74 void Database::setName(const char *name)
  75 {
  76     strcpy(metaData_->dbName_ , name);
  77 }
  78
  79 void Database::setCurrentSize(long size)
  80 {
  81     metaData_->curSize_ = size;
  82 }
  83
  84 void Database::setCurrentPage(Page *page)
  85 {
  86     //metaData_->curPage_ = page;
  87     Mutex::CASL((long*)&metaData_->curPage_, (long)metaData_->curPage_, (long)page);
  88 }
  89
  90 void Database::setFirstPage(Page *page)
  91 {
  92     metaData_->firstPage_ = page;
  93 }
  94
  95 void Database::setMaxSize(long size)
  96 {
  97     metaData_->maxSize_ = size;
  98 }
  99
 100 void Database::setNoOfChunks(int chunks)
 101 {
 102     metaData_->noOfChunks_ = chunks;
 103 }
 104
 105 void Database::setHashIndexChunk(Chunk *ch)
 106 {
 107     metaData_->hashIndexChunk_ = ch;
 108 }
 109
 110 void Database::printDebugMutexInfo()
 111 {
 112    metaData_->dbAllocMutex_.print();
 113    metaData_->ckptMutex_.print();
 114    metaData_->dbTransTableMutex_.print();
 115    metaData_->dbProcTableMutex_.print();
 116    metaData_->dbPrepareStmtMutex_.print();
 117    metaData_->chunkUniqueID_.print();
 118 }
 119
 120 int Database::initAllocDatabaseMutex()
 121 {
 122     return metaData_->dbAllocMutex_.init("allocdb");
 123 }
 124
 125 DbRetVal Database::getAllocDatabaseMutex(bool procAccount)
 126 {
 127     struct timeval timeout, timeval;
 128     timeout.tv_sec = Conf::config.getMutexSecs();
 129     timeout.tv_usec = Conf::config.getMutexUSecs();
 130     int tries=0;
 131     int totalTries = Conf::config.getMutexRetries() *2;
 132     int ret =0;
 133     while (tries < totalTries)
 134     {
 135         ret = metaData_->dbAllocMutex_.getLock(procSlot, procAccount);
 136         if (ret == 0) break;
 137         timeval.tv_sec = timeout.tv_sec;
 138         timeval.tv_usec = timeout.tv_usec;
 139         os::select(0, 0, 0, 0, &timeval);
 140         tries++;
 141     }
 142     if (tries >= totalTries) return ErrLockTimeOut;
 143     return OK;
 144 }
 145
 146 DbRetVal Database::releaseAllocDatabaseMutex(bool procAccount)
 147 {
 148     metaData_->dbAllocMutex_.releaseLock(procSlot, procAccount);
 149     return OK;
 150 }
 151
 152 int Database::initPrepareStmtMutex()
 153 {
 154     return metaData_->dbPrepareStmtMutex_.init("prepstmt");
 155 }
 156
 157 DbRetVal Database::getPrepareStmtMutex(bool procAccount)
 158 {
 159     struct timeval timeout, timeval;
 160     timeout.tv_sec = Conf::config.getMutexSecs();
 161     timeout.tv_usec = Conf::config.getMutexUSecs();
 162     int tries=0;
 163     int totalTries = Conf::config.getMutexRetries() *2;
 164     int ret =0;
 165     while (tries < totalTries)
 166     {
 167         ret = metaData_->dbPrepareStmtMutex_.getLock(procSlot, procAccount);
 168         if (ret == 0) break;
 169         timeval.tv_sec = timeout.tv_sec;
 170         timeval.tv_usec = timeout.tv_usec;
 171         os::select(0, 0, 0, 0, &timeval);
 172         tries++;
 173     }
 174     if (tries >= totalTries) return ErrLockTimeOut;
 175     return OK;
 176
 177 }
 178
 179 DbRetVal Database::releasePrepareStmtMutex(bool procAccount)
 180 {
 181     metaData_->dbPrepareStmtMutex_.releaseLock(procSlot, procAccount);
 182     return OK;
 183 }
 184
 185 int Database::initTransTableMutex()
 186 {
 187     return metaData_->dbTransTableMutex_.init("transtable");
 188 }
 189
 190 DbRetVal Database::getTransTableMutex()
 191 {
 192     struct timeval timeout, timeval;
 193     timeout.tv_sec = Conf::config.getMutexSecs();
 194     timeout.tv_usec = Conf::config.getMutexUSecs();
 195     int tries=0;
 196     int totalTries = Conf::config.getMutexRetries() *2;
 197     int ret =0;
 198     while (tries < totalTries)
 199     {
 200         ret = metaData_->dbTransTableMutex_.getLock(procSlot);
 201         if (ret == 0) break;
 202         timeval.tv_sec = timeout.tv_sec;
 203         timeval.tv_usec = timeout.tv_usec;
 204         os::select(0, 0, 0, 0, &timeval);
 205         tries++;
 206     }
 207     if (tries >= totalTries) return ErrLockTimeOut;
 208     return OK;
 209
 210 }
 211
 212 DbRetVal Database::releaseTransTableMutex()
 213 {
 214     metaData_->dbTransTableMutex_.releaseLock(procSlot);
 215     return OK;
 216 }
 217
 218
 219
 220 int Database::initProcessTableMutex()
 221 {
 222     return metaData_->dbProcTableMutex_.init("proctable");
 223 }
 224
 225 DbRetVal Database::getProcessTableMutex(bool procAccount)
 226 {
 227     struct timeval timeout, timeval;
 228     timeout.tv_sec = Conf::config.getMutexSecs();
 229     timeout.tv_usec = Conf::config.getMutexUSecs();
 230     int tries=0;
 231     int totalTries = Conf::config.getMutexRetries() *2;
 232     int ret =0;
 233     while (tries < totalTries)
 234     {
 235         ret = metaData_->dbProcTableMutex_.getLock(procSlot, procAccount);
 236         if (ret == 0) break;
 237         timeval.tv_sec = timeout.tv_sec;
 238         timeval.tv_usec = timeout.tv_usec;
 239         os::select(0, 0, 0, 0, &timeval);
 240         tries++;
 241     }
 242     if (tries >= totalTries) return ErrLockTimeOut;
 243     return OK;
 244
 245 }
 246
 247 DbRetVal Database::releaseProcessTableMutex(bool procAccount)
 248 {
 249     metaData_->dbProcTableMutex_.releaseLock(procSlot, procAccount);
 250     return OK;
 251 }
 252
 253 int Database::initCheckpointMutex()
 254 {
 255     return metaData_->ckptMutex_.init("checkpoint");
 256 }
 257
 258 DbRetVal Database::getSCheckpointMutex(bool procAccount)
 259 {
 260     struct timeval timeout, timeval;
 261     timeout.tv_sec = Conf::config.getMutexSecs();
 262     timeout.tv_usec = Conf::config.getMutexUSecs();
 263     int tries=0;
 264     int totalTries = Conf::config.getMutexRetries() *2;
 265     int ret =0;
 266     while (tries < totalTries)
 267     {
 268         ret = metaData_->ckptMutex_.getShareLock(procSlot, procAccount);
 269         if (ret == 0) break;
 270         timeval.tv_sec = timeout.tv_sec;
 271         timeval.tv_usec = timeout.tv_usec;
 272         os::select(0, 0, 0, 0, &timeval);
 273         tries++;
 274     }
 275     if (tries >= totalTries) return ErrLockTimeOut;
 276     return OK;
 277
 278 }
 279
 280 DbRetVal Database::getXCheckpointMutex(bool procAccount)
 281 {
 282     struct timeval timeout, timeval;
 283     timeout.tv_sec = Conf::config.getMutexSecs();
 284     timeout.tv_usec = Conf::config.getMutexUSecs();
 285     int tries=0;
 286     int totalTries = Conf::config.getMutexRetries() *2;
 287     int ret =0;
 288     while (tries < totalTries)
 289     {
 290         ret = metaData_->ckptMutex_.getExclusiveLock(procSlot, procAccount);
 291         if (ret == 0) break;
 292         timeval.tv_sec = timeout.tv_sec;
 293         timeval.tv_usec = timeout.tv_usec;
 294         os::select(0, 0, 0, 0, &timeval);
 295         tries++;
 296     }
 297     if (tries >= totalTries) return ErrLockTimeOut;
 298     return OK;
 299
 300 }
 301
 302 DbRetVal Database::releaseCheckpointMutex(bool procAccount)
 303 {
 304     metaData_->ckptMutex_.releaseShareLock(procSlot, procAccount);
 305     return OK;
 306 }
 307
 308 // Gets the free page
 309 // Each page is segmented by PAGE_SIZE, so it checks the pageInfo
 310 // of each page to determine if the page is free
 311 // Algorithm is to scan through the pageInfo objects stored at
 312 // address (db start address + i * PAGE_SIZE) where i = 1..n till end
 313 // database
 314 // But in case of large tuples, pages are merged, so there wont be
 315 // PageInfo object on pages which are merged.
 316 // These pages are skipped by checking the nextPageAfterMerge_ of PageInfo
 317
 318 //NOTE::IMPORTANT::assumes alloc database lock is taken before calling this
 319 Page* Database::getFreePage()
 320 {
 321     Page* page = getFirstPage();
 322     //Page* page = getCurrentPage();
 323     //printDebug(DM_Alloc, "Database::getFreePage firstPage:%x",page);
 324     printDebug(DM_Alloc, "Database::getFreePage currentpage:%x",page);
 325     PageInfo* pageInfo = ((PageInfo*)page);
 326     char* endAddr = ((char*)getMetaDataPtr())  + getMaxSize();
 327     int pageSize = PAGE_SIZE;
 328     bool isEndAddchk=false;
 329     while( 1 == pageInfo->isUsed_)
 330     {
 331         //If any pages are merged to store data larger than PAGE_SIZE
 332         //move to the next page after the merge and check whether it is used
 333         if ( pageInfo->nextPageAfterMerge_ == NULL) {
 334             pageInfo = (PageInfo*)((char*)pageInfo + pageSize);
 335             printDebug(DM_Alloc,"Normal Page:Moving to page:%x",pageInfo);
 336         }
 337         else {
 338             pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
 339             printDebug(DM_Alloc,"Merged Page:Moving to page:%x",pageInfo);
 340         }
 341
 342         if((((char*) pageInfo) + pageSize) >= endAddr )
 343         {
 344             if(!isEndAddchk){
 345                 isEndAddchk=true;
 346                 pageInfo=(PageInfo *)getFirstPage();
 347             }
 348             else
 349                break;
 350         }
 351         if ((char*)pageInfo >= endAddr)
 352         {
 353             //printError(ErrSysInternal,"Invalid address %x",pageInfo);
 354             return NULL;
 355         }
 356
 357     }
 358     if (!isValidAddress(((char*) pageInfo) + pageSize))
 359     {
 360         printError(ErrSysInternal, "Invalid address %x",((char*) pageInfo) + pageSize);
 361         return NULL;
 362     }
 363     setCurrentPage((Page*) pageInfo);
 364     printDebug(DM_Alloc,"Database::getFreePage returning page:%x",pageInfo);
 365     return (Page*) pageInfo ;
 366 }
 367
 368 //Used by tuples more than PAGE_SIZE
 369 //NOTE::IMPORTANT::assumes alloc database lock is taken before calling this
 370 Page* Database::getFreePage(size_t size)
 371 {
 372     Page* page = getFirstPage();
 373     PageInfo* pageInfo = ((PageInfo*)page);
 374     int multiple = size / PAGE_SIZE;
 375     int offset = ((multiple + 1) * PAGE_SIZE);
 376     printDebug(DM_Alloc, "Database::getFreePage firstPage:%x size:%ld",page, size);
 377     char* endAddr = ((char*)getMetaDataPtr())  + getMaxSize();
 378     int pageSize = PAGE_SIZE;
 379     bool isEndAddchk = false;
 380     while(true){
 381         while( 1 == pageInfo->isUsed_)
 382         {
 383             //If any pages are merged to store data larger than PAGE_SIZE
 384             //move to the next page after the merge and check whether it is used
 385             if ( pageInfo->nextPageAfterMerge_ == NULL) {
 386                 pageInfo = (PageInfo*)((char*)pageInfo + pageSize);
 387                 printDebug(DM_Alloc,"Normal Page:Moving to page:%x",pageInfo);
 388             }
 389             else {
 390                 pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
 391                 printDebug(DM_Alloc,"Merged Page:Moving to page:%x",pageInfo);
 392             }
 393             if((((char*) pageInfo) + offset) >= endAddr )
 394             {
 395                 if(!isEndAddchk){
 396                     isEndAddchk=true;
 397                     pageInfo=(PageInfo *)getFirstPage();
 398                 }
 399                 else
 400                    break;
 401             }
 402         }
 403         int i = 0;
 404         PageInfo *pInfo = pageInfo;
 405         if ((((char*)pInfo) + offset) >= endAddr)
 406         {
 407             printError(ErrSysInternal,"Invalid address %x",((char*)pInfo) + offset);
 408             return NULL;
 409         }
 410         for (i = 0; i< multiple + 1; i++)
 411         {
 412             if (1 == pInfo->isUsed_) break;
 413             pInfo = (PageInfo*)((char*)pInfo + pageSize);
 414         }
 415         if ( i == (multiple + 1))  break;
 416         pageInfo = (PageInfo*)((char*)pInfo + pageSize);
 417     }
 418
 419     printDebug(DM_Alloc,"Database::getFreePage returning page:%x",pageInfo);
 420     setCurrentPage((Page*) pageInfo);
 421     return (Page*) pageInfo ;
 422 }
 423
 424 void Database::printStatistics()
 425 {
 426     Page* page = getFirstPage();
 427     PageInfo* pageInfo = ((PageInfo*)page);
 428     int usedPageCount =0, usedMergedPageCount =0, totalPages=0;
 429     int totalDirtyPages=0;
 430     printf("<DatabaseStatistics>\n");
 431     printf("  <Database Name>  %s </Database Name>\n", getName());
 432     printf("  <Max Size> %ld </Max Size>\n", getMaxSize());
 433     printf("  <First Page> %x </First Page>\n", getFirstPage());
 434     while(isValidAddress((char*) pageInfo))
 435     {
 436         if (pageInfo == NULL) break;
 437         //if (pageInfo > getCurrentPage()) break;
 438         if (1 == pageInfo->isUsed_) {
 439            if ( pageInfo->nextPageAfterMerge_ == NULL) {
 440               if (BITSET(pageInfo->flags, IS_DIRTY)) totalDirtyPages++;
 441               pageInfo = (PageInfo*)((char*)pageInfo + PAGE_SIZE);
 442               usedPageCount++; totalPages++;
 443               printDebug(DM_Alloc, "Normal Page:Moving to page:%x\n",pageInfo);
 444               continue;
 445            }
 446            else {
 447               if (BITSET(pageInfo->flags, IS_DIRTY)) totalDirtyPages++;
 448               pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
 449               usedMergedPageCount++; totalPages++;
 450               printDebug(DM_Alloc,"Merged Page:Moving to page:%x\n",pageInfo);
 451               continue;
 452            }
 453         } else if (BITSET(pageInfo->flags, IS_DIRTY)) totalDirtyPages++;
 454         pageInfo = (PageInfo*)((char*)pageInfo + PAGE_SIZE);
 455         printDebug(DM_Alloc,"Normal Page not used:Moving to page:%x\n",pageInfo);
 456         totalPages++;
 457     }
 458     printf("  <Total Pages> %d </Total Pages>\n", totalPages);
 459     if (Conf::config.useDurability())
 460         printf("  <Dirty Pages> %d </Dirty Pages>\n", totalDirtyPages);
 461     printf("  <Used Normal Pages> %d </Used Normal Pages>\n", usedPageCount);
 462     printf("  <Used Merged Pages> %d </Used Merged Pages>\n", usedMergedPageCount);
 463     printf("  <Chunks Used> %d </Chunks Used>\n", getNoOfChunks());
 464     printf("</DatabaseStatistics>\n");
 465
 466     return ;
 467 }
 468
 469
 470 //called only in case of system database to create and initialize the chunk
 471 //information
 472 DbRetVal Database::createSystemDatabaseChunk(AllocType type, size_t size, int id)
 473 {
 474
 475     Chunk *chunk;
 476     if (-1 == id )
 477     {
 478         printError(ErrSysFatal, "Database ID corrupted");
 479         return ErrSysFatal;
 480     }
 481     chunk = getSystemDatabaseChunk(id);
 482
 483     chunk->setChunkNameForSystemDB(id);
 484
 485     if (FixedSizeAllocator == type) chunk->setSize(size);
 486     //getDatabaseMutex();
 487     if (chunk->allocSize_ > PAGE_SIZE)
 488         chunk->curPage_ = getFreePage(chunk->allocSize_);
 489     else
 490          chunk->curPage_ = getFreePage();
 491     if ( chunk->curPage_ == NULL)
 492     {
 493        //releaseDatabaseMutex();
 494        printError(ErrNoMemory, "No free pages in database: Database full");
 495        return ErrNoMemory;
 496     }
 497
 498     chunk->firstPage_ = chunk->curPage_;
 499     PageInfo* firstPageInfo = ((PageInfo*)chunk->firstPage_);
 500     chunk->setChunkID(id);
 501     chunk->setAllocType(type);
 502     chunk->initMutex(id);
 503     printDebug(DM_Database, "Creating System Database Chunk:%d Size:%d",id, chunk->allocSize_);
 504     if (chunk->allocSize_ > PAGE_SIZE)
 505     {
 506         int multiple = os::floor(chunk->allocSize_ / PAGE_SIZE);
 507         int offset = ((multiple + 1) * PAGE_SIZE);
 508         firstPageInfo->setPageAsUsed(offset);
 509     }
 510     else {
 511         firstPageInfo->setPageAsUsed(chunk->allocSize_);
 512     }
 513
 514     if (0 == size)
 515     {
 516         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
 517         varInfo->isUsed_ = 0;
 518         varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
 519
 520     }
 521     incrementChunk();
 522     //releaseDatabaseMutex();
 523     return OK;
 524 }
 525
 526 //This is never called currently. If situation arises will be coded later.
 527 DbRetVal Database::deleteSystemDatabaseChunk(int id)
 528 {
 529
 530     Chunk *chunk = getSystemDatabaseChunk(id);
 531     chunk->setChunkID(-1);
 532     chunk->setSize(0);
 533     chunk->setAllocType(UnknownAllocator);
 534     //TODO::
 535     //chunk->pageList_
 536     //walk though the pageList ptr and get all the page pointers
 537     //then free all the pages used to store this by setting the
 538     //start of page to notused
 539     chunk->firstPage_ = NULL;
 540     chunk->curPage_ = NULL;
 541     decrementChunk();
 542     return OK;
 543 }
 544
 545
 546 void Database::createAllCatalogTables()
 547 {
 548     //These are special chunks which hold catalog tables and other information
 549     //
 550     // chunk id 0 ->userChunkTable
 551     // chunk id 1 ->lockBucketHash
 552     // chunk id 2 ->lockTable
 553     //
 554     // chunk id 10->DATABASE
 555     // chunk id 11->USER
 556     // chunk id 12->TABLE
 557     // chunk id 13->FIELD
 558     // chunk id 14->ACCESS
 559
 560     createSystemTables();
 561     createMetaDataTables();
 562 }
 563
 564 void Database::createSystemTables()
 565 {
 566     createSystemDatabaseChunk(FixedSizeAllocator,
 567                                   sizeof(Chunk), UserChunkTableId);
 568     createSystemDatabaseChunk(FixedSizeAllocator,
 569                                   sizeof(Bucket) * LOCK_BUCKET_SIZE,
 570                                   LockTableHashBucketId);
 571     createSystemDatabaseChunk(FixedSizeAllocator,
 572                                   sizeof(Mutex)* LOCK_BUCKET_SIZE,
 573                                   LockTableMutexId);
 574     createSystemDatabaseChunk(FixedSizeAllocator,
 575                                   sizeof(LockHashNode), LockTableId);
 576     createSystemDatabaseChunk(FixedSizeAllocator,
 577                                   sizeof(TransHasNode), TransHasTableId);
 578
 579     createSystemDatabaseChunk(VariableSizeAllocator,
 580                                   0, UndoLogTableID);
 581 }
 582
 583 void Database::createMetaDataTables()
 584 {
 585     createSystemDatabaseChunk(FixedSizeAllocator,
 586                                   sizeof(CDATABASEFILE), DatabaseTableId);
 587     createSystemDatabaseChunk(FixedSizeAllocator,
 588                                   sizeof(CUSER), UserTableId);
 589     createSystemDatabaseChunk(FixedSizeAllocator,
 590                                   sizeof(CTABLE), TableTableId);
 591     createSystemDatabaseChunk(FixedSizeAllocator,
 592                                   sizeof(CFIELD), FieldTableId);
 593     createSystemDatabaseChunk(FixedSizeAllocator,
 594                                   sizeof(CACCESS), AccessTableId);
 595     createSystemDatabaseChunk(FixedSizeAllocator,
 596                                   sizeof(CINDEX), IndexTableId);
 597     createSystemDatabaseChunk(FixedSizeAllocator,
 598                                   sizeof(CINDEXFIELD), IndexFieldTableId);
 599     createSystemDatabaseChunk(FixedSizeAllocator,
 600                                   sizeof(CFK), ForeignKeyTableId);
 601     createSystemDatabaseChunk(FixedSizeAllocator,
 602                                   sizeof(CFKFIELD), ForeignKeyFieldTableId);
 603 }
 604
 605 //used in case of system database
 606 Chunk* Database::getSystemDatabaseChunk(int id)
 607 {
 608     size_t offset = os::alignLong(sizeof (DatabaseMetaData)) +
 609                         id * sizeof (Chunk);
 610     return (Chunk*)(((char*) metaData_) +  offset);
 611 }
 612
 613
 614 //used in case of system database
 615 Transaction* Database::getSystemDatabaseTrans(int slot)
 616 {
 617     size_t offset = os::alignLong(sizeof (DatabaseMetaData)) +
 618                     os::alignLong(MAX_CHUNKS * sizeof (Chunk)) +
 619                         slot  * sizeof (Transaction);
 620     return (Transaction*)(((char*) metaData_) +  offset);
 621 }
 622
 623 bool Database::isValidAddress(void* addr)
 624 {
 625     if ((char*) addr >= ((char*)getMetaDataPtr())  + getMaxSize())
 626         return false;
 627     else
 628         return true;
 629 }
 630
 631 //should be called only on system database
 632 void* Database::allocLockHashBuckets()
 633 {
 634     Chunk *chunk = getSystemDatabaseChunk(LockTableHashBucketId);
 635     DbRetVal rv=OK;
 636     void *ptr = chunk->allocate(this, &rv);
 637     if (NULL == ptr)
 638     {
 639         printError(ErrNoMemory, "Chunk Allocation failed for lock hash bucket catalog table");
 640     }
 641     return ptr;
 642 }
 643
 644 Bucket* Database::getLockHashBuckets()
 645 {
 646     Chunk *tChunk = getSystemDatabaseChunk(LockTableHashBucketId);
 647     ChunkIterator iter = tChunk->getIterator();
 648     return (Bucket*)iter.nextElement();
 649 }
 650
 651 void Database::setUniqueChunkID(int id)
 652 {
 653     (metaData_->chunkUniqueID_).setID(id);
 654 }
 655
 656 int Database::getUniqueIDForChunk()
 657 {
 658     return ((metaData_->chunkUniqueID_).getID());
 659 }
 660
 661 DbRetVal Database::recoverMutex(Mutex *mut)
 662 {
 663     //TODO: operations need to be undone before recovering the mutex.
 664     mut->recoverMutex();
 665     return OK;
 666 }
 667