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