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<TableConfig.h>
  29 #include<Process.h>
  30
  31
  32 DatabaseManagerImpl::~DatabaseManagerImpl()
  33 {
  34     //Note:Databases are closed by the session interface
  35     delete tMgr_;
  36     delete lMgr_;
  37 }
  38
  39 void DatabaseManagerImpl::createLockManager()
  40 {
  41     lMgr_ = new LockManager(systemDatabase_);
  42     return;
  43 }
  44
  45 void DatabaseManagerImpl::createTransactionManager()
  46 {
  47
  48     tMgr_ = new TransactionManager();
  49     tMgr_->setFirstTrans(systemDatabase_->getSystemDatabaseTrans(0));
  50     return;
  51 }
  52 void DatabaseManagerImpl::setProcSlot()
  53 {
  54     systemDatabase_->setProcSlot(procSlot);
  55     db_->setProcSlot(procSlot);
  56 }
  57 DbRetVal DatabaseManagerImpl::openSystemDatabase()
  58 {
  59     DbRetVal rv = openDatabase(SYSTEMDB);
  60     if (rv != OK) return rv;
  61     systemDatabase_ = db_;
  62     db_ = NULL;
  63     printDebug(DM_Database, "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     return OK;
  77 }
  78
  79 DbRetVal DatabaseManagerImpl::createDatabase(const char *name, size_t size)
  80 {
  81     bool isMmapNeeded = Conf::config.useMmap();
  82 /*
  83     if (isMmapNeeded && !Conf::config.useDurability()) {
  84         printError(ErrBadArg, "If MMAP is set to true. Durability must be true.");
  85         return ErrBadArg;
  86     }
  87 */
  88     int fd = -1;
  89     char cmd[1024];
  90     char dbMapFile[MAX_FILE_LEN];
  91     struct stat st;
  92     long fixAddr = 399998976L;
  93     bool firstTimeServer = false;
  94     if (NULL != db_ )
  95     {
  96         printError(ErrAlready, "Database is already created");
  97         return ErrAlready;
  98     }
  99     caddr_t rtnAddr = (caddr_t) NULL;
 100     void *mapAddr = NULL;
 101     shared_memory_id shm_id = 0;
 102
 103     char *startaddr = (char*)Conf::config.getMapAddress();
 104     shared_memory_key key = 0;
 105     if (0 == strcmp(name, SYSTEMDB))
 106     {
 107
 108         key = Conf::config.getSysDbKey();
 109     }
 110     else
 111     {
 112         startaddr = startaddr + Conf::config.getMaxSysDbSize();
 113         key = Conf::config.getUserDbKey();
 114     }
 115     if (!isMmapNeeded || (isMmapNeeded && 0 == strcmp(name, SYSTEMDB))) {
 116         shm_id = os::shm_create(key, size, 0660);
 117         if (-1 == shm_id) {
 118             if (errno == EEXIST)
 119                 printError(ErrOS, "Shared Memory already exists");
 120             printError(ErrOS, "Shared memory create failed");
 121             shm_id = os::shm_open(Conf::config.getSysDbKey(), 100, 0660);
 122             os::shmctl(shm_id, IPC_RMID);
 123             delete systemDatabase_;
 124             systemDatabase_ = NULL;
 125             return ErrOS;
 126         }
 127     } else {
 128         //switch the checkpoint
 129         if (Database::getCheckpointID() == 0)
 130             Database::setCheckpointID(1);
 131         else
 132             Database::setCheckpointID(0);
 133         int chkptID=Database::getCheckpointID();
 134
 135
 136         sprintf(dbMapFile, "%s/db.chkpt.data%d", Conf::config.getDbFile(), chkptID);
 137         /*if (FILE *file = fopen(dbMapFile, "r")) {
 138             fclose(file);
 139             sprintf(cmd, "cp %s %s/db.chkpt.data", dbMapFile, Conf::config.getDbFile());
 140             int ret = system(cmd);
 141             if (ret != 0) {
 142                 printError(ErrOS, "could not copy data file to map file");
 143                 return ErrOS;
 144             }
 145         }
 146         sprintf(dbMapFile, "%s/db.chkpt.data", Conf::config.getDbFile());
 147         */
 148
 149
 150         fd = open(dbMapFile, O_CREAT | O_RDWR, 0660);
 151         if (-1 == fd) {
 152             printError(ErrOS, "Mmap file could not be opened");
 153             return ErrOS;
 154         }
 155         if(fstat(fd, &st) == -1) {
 156             printf("Unable to retrieve the db File data\n");
 157             close(fd);
 158             db_->setChkptfd(-1);
 159             return ErrOS;
 160         }
 161         if (st.st_size == 0 || st.st_size < size) {
 162             firstTimeServer = true;
 163             off_t flSize = lseek(fd, size - 1, SEEK_SET);
 164         }
 165         char *a = "0";
 166         int wSize = write(fd, a, 1);
 167         mapAddr = os::mmap((void *)(fixAddr + Conf::config.getMaxSysDbSize()), size, PROT_READ | PROT_WRITE, MAP_FIXED | MAP_SHARED, fd, 0);
 168         rtnAddr = (caddr_t) mapAddr;
 169         printDebug(DM_Database, "Mapped db file address = %x", mapAddr);
 170     }
 171     void *shm_ptr = NULL;
 172     if (!isMmapNeeded || isMmapNeeded && 0 == strcmp(name, SYSTEMDB)) {
 173         shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 174         rtnAddr  = (caddr_t) shm_ptr;
 175         if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
 176         {
 177             printError(ErrOS, "Shared memory attach returned -ve value %d", rtnAddr);
 178             return ErrOS;
 179         }
 180     }
 181     db_ = new Database();
 182     printDebug(DM_Database, "Creating database:%s",name);
 183
 184     /*if (!isMmapNeeded || isMmapNeeded && 0 == strcmp(name, SYSTEMDB)) {
 185         memset(shm_ptr, 0, size );
 186     }*/
 187
 188     //TODO:for user database do not have transtable and processtable mutex
 189     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 190     db_->setDatabaseID(1);
 191     db_->setName(name);
 192     db_->setMaxSize(size);
 193     db_->setNoOfChunks(0);
 194     db_->initAllocDatabaseMutex();
 195     db_->initTransTableMutex();
 196     db_->initCheckpointMutex();
 197     db_->initProcessTableMutex();
 198     db_->initPrepareStmtMutex();
 199     db_->setUniqueChunkID(100);
 200     //compute the first page after book keeping information
 201     size_t offset = os::alignLong(sizeof (DatabaseMetaData));
 202     //Only for system db chunk array, trans array and proc array will be there
 203     if (0 == strcmp(name, SYSTEMDB))
 204     {
 205         offset = offset + os::alignLong( MAX_CHUNKS  * sizeof (Chunk));
 206         offset = offset + os::alignLong( Conf::config.getMaxProcs()   * sizeof(Transaction));
 207         offset = offset + os::alignLong( Conf::config.getMaxProcs() * sizeof(ThreadInfo));
 208     }
 209     int multiple = os::floor(offset / PAGE_SIZE);
 210     char *curPage = (((char*)rtnAddr) + ((multiple + 1) * PAGE_SIZE));
 211
 212     db_->setCurrentPage(curPage);
 213     db_->setFirstPage(curPage);
 214
 215     if (0 == strcmp(name, SYSTEMDB)) return OK;
 216
 217     /*Allocate new chunk to store hash index nodes
 218     Chunk *chunkInfo = createUserChunk(sizeof(HashIndexNode));
 219     if (NULL == chunkInfo)
 220     {
 221         printError(ErrSysInternal, "Failed to allocate hash index nodes chunk");
 222         return ErrSysInternal;
 223     }
 224     printDebug(DM_Database, "Creating Chunk for storing Hash index nodes %x",
 225                                                                   chunkInfo);
 226
 227     db_->setHashIndexChunk(chunkInfo);*/
 228     return OK;
 229 }
 230
 231 DbRetVal DatabaseManagerImpl::deleteDatabase(const char *name)
 232 {
 233     shared_memory_id shm_id = 0;
 234     if (0 == strcmp(name, SYSTEMDB))
 235     {
 236         shm_id = os::shm_open(Conf::config.getSysDbKey(), 100, 0660);
 237         os::shmctl(shm_id, IPC_RMID);
 238                 delete systemDatabase_;
 239                 systemDatabase_ = NULL;
 240     } else {
 241         shm_id = os::shm_open(Conf::config.getUserDbKey(), 100, 0660);
 242         os::shmctl(shm_id, IPC_RMID);
 243                 delete db_;
 244                 db_ = NULL;
 245     }
 246     return OK;
 247 }
 248
 249 DbRetVal DatabaseManagerImpl::openDatabase(const char *name)
 250 {
 251     bool isMmapNeeded = Conf::config.useMmap();
 252     char dbMapFile[1024];
 253     int fd = -1;
 254     long size = Conf::config.getMaxSysDbSize();
 255     char *startaddr = (char*)Conf::config.getMapAddress();
 256     long fixAddr = 399998976L;
 257     if (0 == strcmp(name , SYSTEMDB))
 258     {
 259         if (NULL !=systemDatabase_)
 260         {
 261             printError(ErrAlready, "System Database already open");
 262             return ErrAlready;
 263         }
 264     }
 265     else
 266     {
 267         if (NULL ==systemDatabase_)
 268         {
 269             printError(ErrNotOpen, "System Database not open");
 270             return ErrNotOpen;
 271         }
 272         size = Conf::config.getMaxDbSize();
 273         startaddr = startaddr +  Conf::config.getMaxSysDbSize();
 274         fixAddr += Conf::config.getMaxSysDbSize();
 275     }
 276     if (NULL != db_)
 277     {
 278         printError(ErrAlready, "User Database already open");
 279         return ErrAlready;
 280     }
 281     //system db should be opened before user database files
 282     caddr_t rtnAddr = (caddr_t) NULL;
 283
 284     shared_memory_id shm_id = 0;
 285     shared_memory_key key = 0;
 286
 287     if (0 == strcmp(name, SYSTEMDB))
 288         key = Conf::config.getSysDbKey();
 289     else
 290        key = Conf::config.getUserDbKey();
 291
 292
 293     void *shm_ptr = NULL;
 294     void *mapAddr = NULL;
 295     bool firstThread = false;
 296     if ( ( ProcessManager::noThreads == 0 && 0 == strcmp(name, SYSTEMDB) ||
 297           ProcessManager::noThreads == 1 && 0 != strcmp(name, SYSTEMDB) ) )
 298    {
 299         if(isMmapNeeded && 0 != strcmp(name, SYSTEMDB)){
 300             //: Attach to Map File
 301             int curChkptID = Database::getCheckpointID();
 302             sprintf(dbMapFile, "%s/db.chkpt.data%d", Conf::config.getDbFile(),
 303                                                      curChkptID);
 304             fd = open(dbMapFile, O_RDWR, 0660);
 305             if (-1 == fd) {
 306                 printError(ErrOS, "Mmap file could not be opened");
 307                 return ErrOS;
 308             }
 309             mapAddr = os::mmap((void *)fixAddr, size, PROT_READ | PROT_WRITE, MAP_FIXED | MAP_SHARED, fd, 0);
 310
 311             shm_ptr= (caddr_t) mapAddr;
 312             printDebug(DM_Database, "Mapped db file address = %x", mapAddr);
 313         }else
 314         {
 315             shm_id = os::shm_open(key, size, 0660);
 316             if (shm_id == -1 )
 317             {
 318                 printError(ErrOS, "Shared memory open failed");
 319                 return ErrOS;
 320             }
 321             shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 322         }
 323         if (0 == strcmp(name, SYSTEMDB))
 324         {
 325             firstThread = true;
 326             ProcessManager::sysAddr = (char*) shm_ptr;
 327         }
 328         else
 329         {
 330             ProcessManager::usrAddr = (char*) shm_ptr;
 331         }
 332     } else {
 333         if (0 == strcmp(name, SYSTEMDB))
 334             shm_ptr = ProcessManager::sysAddr;
 335         else shm_ptr = ProcessManager::usrAddr;
 336     }
 337
 338
 339         rtnAddr  = (caddr_t) shm_ptr;
 340 #if defined (x86_64)
 341     if (rtnAddr < 0 || shm_ptr == (char*)0xffffffffffffffff)
 342     {
 343         printError(ErrOS, "Shared memory attach returned -ve value %x %d", shm_ptr, errno);
 344         return ErrOS;
 345     }
 346 #else
 347     if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
 348     {
 349         printError(ErrOS, "Shared memory attach returned -ve value %x %d", shm_ptr, errno);
 350         return ErrOS;
 351     }
 352 #endif
 353
 354     db_ = new Database();
 355     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 356     db_->setChkptfd(fd);
 357
 358     if (firstThread) ProcessManager::systemDatabase = db_;
 359
 360     printDebug(DM_Database, "Opening database: %s", name);
 361     return OK;
 362 }
 363
 364 DbRetVal DatabaseManagerImpl::closeDatabase()
 365 {
 366
 367     if (NULL == db_)
 368     {
 369         //Database is already closed
 370         return OK;
 371     }
 372     printDebug(DM_Database, "Closing database: %s",(char*)db_->getName());
 373     //check if this is the last thread to be deregistered
 374     int ret =0;// ProcessManager::mutex.getLock(-1, false);
 375     //If you are not getting lock ret !=0, it means somebody else is there.
 376     //he will close the database.
 377     if (ret == 0) {
 378     if (ProcessManager::noThreads == 0 && 0 == strcmp((char*)db_->getName(), SYSTEMDB)
 379        || ProcessManager::noThreads == 1 && 0 != strcmp((char*)db_->getName(),  SYSTEMDB) )  {
 380                 os::shm_detach((char*)db_->getMetaDataPtr());
 381             }
 382     }
 383   //  ProcessManager::mutex.releaseLock(-1, false);
 384     delete db_;
 385     db_ = NULL;
 386     return OK;
 387 }
 388 //Assumes that system database mutex is taken before calling this.
 389 Chunk* DatabaseManagerImpl::createUserChunk(size_t size)
 390 {
 391     //Allocate new node in system database to store
 392     Chunk *chunk = getSystemTableChunk(UserChunkTableId);
 393     DbRetVal rv = OK;
 394     void *ptr = chunk->allocate(systemDatabase_, &rv);
 395     if (NULL == ptr)
 396     {
 397          printError(rv, "Allocation failed for User chunk catalog table");
 398          return NULL;
 399     }
 400     Chunk *chunkInfo = (Chunk*)ptr;
 401     chunkInfo->initMutex();
 402     if (0 != size) chunkInfo->setSize(size);
 403     if (chunkInfo->allocSize_ > PAGE_SIZE)
 404         chunkInfo->curPage_ = db_->getFreePage(chunkInfo->allocSize_);
 405     else
 406         chunkInfo->curPage_ = db_->getFreePage();
 407     if ( NULL == chunkInfo->curPage_)
 408     {
 409         chunkInfo->destroyMutex();
 410         chunk->free(db_, ptr);
 411         printError(ErrNoMemory, "Database full: No space to allocate from database");
 412         return NULL;
 413     }
 414     PageInfo* firstPageInfo = ((PageInfo*)chunkInfo->curPage_);
 415     if (chunkInfo->allocSize_ > PAGE_SIZE)
 416     {
 417         int multiple = os::floor(chunkInfo->allocSize_ / PAGE_SIZE);
 418         int offset = ((multiple + 1) * PAGE_SIZE);
 419         firstPageInfo->setPageAsUsed(offset);
 420     }
 421     else
 422     {
 423         firstPageInfo->setPageAsUsed(chunkInfo->allocSize_);
 424         char *data = ((char*)firstPageInfo) + sizeof(PageInfo);
 425         *(InUse*)data =0;
 426     }
 427     if (0 == size)
 428     {
 429         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
 430         varInfo->isUsed_ = 0;
 431         varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
 432
 433     }
 434     chunkInfo->firstPage_ = chunkInfo->curPage_;
 435
 436     if (0 == size)
 437         chunkInfo->setAllocType(VariableSizeAllocator);
 438     else
 439         chunkInfo->setAllocType(FixedSizeAllocator);
 440
 441     //TODO::Generate chunkid::use tableid
 442     chunkInfo->setChunkID(db_->getUniqueIDForChunk());
 443     db_->incrementChunk();
 444     chunkInfo->setPageDirty(firstPageInfo);
 445     printDebug(DM_Database, "Creating new User chunk chunkID:%d size: %d firstPage:%x",
 446                                -1, chunkInfo->allocSize_, firstPageInfo);
 447
 448     return chunkInfo;
 449 }
 450
 451 //Assumes that system database mutex is taken before calling this.
 452 DbRetVal DatabaseManagerImpl::deleteUserChunk(Chunk *chunk)
 453 {
 454     //Go to the pages and set them to notUsed
 455     Page *page = chunk->firstPage_;
 456     PageInfo* pageInfo = ((PageInfo*)page);
 457     //Here...sure that atleast one page will be there even no tuples
 458     //are inserted.so not checking if pageInfo == NULL
 459     while( pageInfo->nextPage_ != NULL)
 460     {
 461         PageInfo *prev = pageInfo;
 462         pageInfo = (PageInfo*)(pageInfo->nextPage_);
 463         //sets pageInfo->isUsed_ = 0 and pageInfo->hasFreeSpace_ = 0
 464         //and initializes the page content to zero
 465         if(NULL == pageInfo->nextPageAfterMerge_){
 466             os::memset(prev, 0, PAGE_SIZE);
 467             SETBIT(prev->flags, IS_DIRTY);
 468
 469         }
 470         else
 471         {
 472             int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 473             char *iter = (char*)prev, *end=(char*)pageInfo->nextPageAfterMerge_;
 474             os::memset(prev, 0, size);
 475             //set dirty bit for all pages in merged pages
 476             while(iter <= end)
 477             {
 478                 PageInfo *info = (PageInfo*) iter;
 479                 SETBIT(info->flags, IS_DIRTY);
 480                 iter = iter  + PAGE_SIZE;
 481             }
 482         }
 483         printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, prev);
 484     }
 485     //The above loop wont execute for the last page
 486     //and for the case where table has only one page
 487     if(NULL == pageInfo->nextPageAfterMerge_) {
 488         os::memset(pageInfo, 0, PAGE_SIZE);
 489         SETBIT(pageInfo->flags, IS_DIRTY);
 490     }
 491     else
 492     {
 493         int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 494         char *iter = (char*)pageInfo, *end=(char*)pageInfo->nextPageAfterMerge_;
 495         os::memset(pageInfo, 0, size);
 496         //set dirty bit for all pages in merged pages
 497         while(iter <= end)
 498         {
 499             PageInfo *info = (PageInfo*) iter;
 500             SETBIT(info->flags, IS_DIRTY);
 501             iter = iter  + PAGE_SIZE;
 502         }
 503     }
 504     printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, pageInfo);
 505     chunk->chunkID_ = -1;
 506     chunk->allocSize_ = 0;
 507     chunk->curPage_ = NULL;
 508     chunk->firstPage_ = NULL;
 509     chunk->destroyMutex();
 510     db_->decrementChunk();
 511     Chunk *userChunk = getSystemTableChunk(UserChunkTableId);
 512     userChunk->free(systemDatabase_,chunk);
 513     printDebug(DM_Database,"deleting user chunk:%x",chunk);
 514     return OK;
 515 }
 516
 517 //-1 -> Unable to create chunk. No memory
 518 //-2 -> Unable to update the catalog tables
 519 DbRetVal DatabaseManagerImpl::createTable(const char *name, TableDef &def)
 520 {
 521     DbRetVal rv = OK;
 522     if (!Util::isIdentifier((char*)name)) {
 523         printError(ErrBadArg, "Invalid character for index name");
 524         return ErrBadArg;
 525     }
 526     int fldCount = def.getFieldCount();
 527     if(0==fldCount)
 528     {
 529         printError(ErrNotExists,"Table can't be created without Field");
 530         return ErrNotExists;
 531     }
 532
 533     //If total field count is less than 32, then 1 integer is used to store all null
 534     //information, if it is more then 1 char is used to store null information
 535     //of each field
 536     //This is to done to reduce cpu cycles for small tables
 537     int addSize = 0;
 538     if (fldCount < 31) addSize = 4; else addSize = os::align(fldCount);
 539     size_t sizeofTuple = os::alignLong(def.getTupleSize()+addSize);
 540     rv = systemDatabase_->getXCheckpointMutex();
 541     if (OK != rv ) {
 542         printError(rv, "Unable to get Database mutex");
 543         return rv;
 544     }
 545
 546     void *tptr =NULL;
 547     void *chunk = NULL;
 548     void *vcchunk = NULL;
 549
 550     //check whether table already exists
 551     CatalogTableTABLE cTable(systemDatabase_);
 552     cTable.getChunkAndTblPtr(name, chunk, tptr, vcchunk);
 553     if (NULL != tptr)
 554     {
 555         systemDatabase_->releaseCheckpointMutex();
 556         printError(ErrAlready, "Table %s already exists", name);
 557         return ErrAlready;
 558     }
 559
 560     //create a chunk to store the tuples
 561     Chunk *ptr = createUserChunk(sizeofTuple);
 562     if (NULL  == ptr)
 563     {
 564         systemDatabase_->releaseCheckpointMutex();
 565         printError(ErrNoResource, "Unable to create user chunk");
 566         return ErrNoResource;
 567     }
 568     printDebug(DM_Database,"Created UserChunk:%x", ptr);
 569     ptr->setChunkName(name);
 570     //add row to TABLE
 571     int tblID = ((Chunk*)ptr)->getChunkID();
 572
 573     //check whether varchar is present in table
 574     FieldIterator fiter = def.getFieldIterator();
 575     bool isVarcharPresent = def.isVarcharPresentInSchema(fiter);
 576     Chunk *vcptr = NULL;
 577     if (isVarcharPresent) {
 578         //creat chunk to store varchar values
 579         vcptr = createUserChunk();
 580         if (NULL  == vcptr)
 581         {
 582             deleteUserChunk(ptr);
 583             systemDatabase_->releaseCheckpointMutex();
 584             printError(ErrNoResource, "Unable to create user chunk for varchar");
 585             return ErrNoResource;
 586         }
 587         printDebug(DM_Database,"Created UserChunk for Varchar:%x", vcptr);
 588         vcptr->setChunkName(name);
 589     }
 590     rv = cTable.insert(name, tblID, sizeofTuple,
 591                                    def.getFieldCount(), ptr, tptr, vcptr);
 592     if (OK != rv)
 593     {
 594         deleteUserChunk(ptr);
 595         if (vcptr) deleteUserChunk(vcptr);
 596         systemDatabase_->releaseCheckpointMutex();
 597         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 598         return ErrSysInternal;
 599     }
 600     printDebug(DM_Database,"Inserted into TABLE:%s",name);
 601     //add rows to FIELD
 602     FieldIterator iter = def.getFieldIterator();
 603     CatalogTableFIELD cField(systemDatabase_);
 604     rv = cField.insert(iter, tblID ,tptr);
 605     if (OK != rv)
 606     {
 607         deleteUserChunk(ptr);
 608         if (vcptr) deleteUserChunk(vcptr);
 609         void *cptr, *ttptr;//Dummy as remove below needs both these OUT params
 610         cTable.remove(name, cptr, ttptr);
 611         systemDatabase_->releaseCheckpointMutex();
 612         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 613         return ErrSysInternal;
 614     }
 615     printDebug(DM_Database,"Inserted into FIELD:%s",name);
 616     systemDatabase_->releaseCheckpointMutex();
 617     printDebug(DM_Database,"Table Created:%s",name);
 618     logFinest(Conf::logger, "Table Created %s" , name);
 619     return OK;
 620 }
 621 DbRetVal DatabaseManagerImpl::renameTable(const char *oldName,const char *newName)
 622 {
 623     void *chunk = NULL;
 624     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 625     if (OK != rv) {
 626         printError(ErrSysInternal, "Unable to get database mutex for rename table");
 627         return ErrSysInternal;
 628     }
 629     CatalogTableTABLE cTable(systemDatabase_);
 630     rv = cTable.renameTable(oldName,newName);
 631     if (OK != rv) {
 632         printError(ErrSysInternal, "Unable to rename table");
 633         systemDatabase_->releaseCheckpointMutex();
 634         return ErrSysInternal;
 635     }
 636     systemDatabase_->releaseCheckpointMutex();
 637     return OK;
 638 }
 639
 640 DbRetVal DatabaseManagerImpl::renameField(const char *tableName,const char *oldName,const char *newName)
 641 {
 642    DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 643     if (OK != rv) {
 644         printError(ErrSysInternal, "Unable to get database mutex for rename table");
 645         return ErrSysInternal;
 646     }
 647     CatalogTableFIELD fTable(systemDatabase_);
 648     rv = fTable.renameField(tableName, oldName, newName);
 649     if (OK != rv) {
 650         printError(ErrSysInternal, "Unable to rename table");
 651         systemDatabase_->releaseCheckpointMutex();
 652         return ErrSysInternal;
 653     }
 654     systemDatabase_->releaseCheckpointMutex();
 655     return rv;
 656 }
 657
 658 //TODO::If any operation fails in between, then we may have some
 659 //dangling tuples, say we have have rows in INDEX table
 660 //which will not have any corresponding entries in TABLE
 661 //CHANGE the sequence so that it deletes from the bottom as
 662 //opposed to start from top as is written now
 663 DbRetVal DatabaseManagerImpl::dropTable(const char *name)
 664 {
 665     void *chunk = NULL;
 666     void *tptr =NULL;
 667     void *vcchunk = NULL;
 668     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 669     if (OK != rv) {
 670         printError(ErrSysInternal, "Unable to get database mutex");
 671         return ErrSysInternal;
 672     }
 673     //remove the entry in TABLE
 674     CatalogTableTABLE cTable(systemDatabase_);
 675     rv = cTable.getChunkAndTblPtr(name, chunk, tptr, vcchunk);
 676     if (OK != rv) {
 677         systemDatabase_->releaseCheckpointMutex();
 678         printError(ErrSysInternal, "Table %s does not exist", name);
 679         return ErrSysInternal;
 680     }
 681     CatalogTableFK cFK(systemDatabase_);
 682     int noOfRelation =cFK.getNumFkTable(tptr);
 683     if(noOfRelation)
 684     {
 685         printError(ErrSysInternal, "Unable to drop table due to relation exist.Drop child table...");
 686         systemDatabase_->releaseCheckpointMutex();
 687         return ErrSysInternal;
 688     }
 689     rv = lMgr_->getExclusiveLock(chunk, NULL);
 690     if (rv !=OK)
 691     {
 692         systemDatabase_->releaseCheckpointMutex();
 693         printError(ErrLockTimeOut, "Unable to acquire exclusive lock on the table\n");
 694         return rv;
 695     }
 696
 697     rv = cTable.remove(name, chunk, tptr);
 698     if (OK != rv) {
 699         systemDatabase_->releaseCheckpointMutex();
 700         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 701         return ErrSysInternal;
 702     }
 703     printDebug(DM_Database,"Deleted from TABLE:%s",name);
 704
 705     //remove the entries in the FIELD table
 706     CatalogTableFIELD cField(systemDatabase_);
 707     rv = cField.remove(tptr);
 708     if (OK != rv) {
 709         systemDatabase_->releaseCheckpointMutex();
 710         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 711         return ErrSysInternal;
 712     }
 713     printDebug(DM_Database,"Deleted from FIELD:%s",name);
 714
 715     rv = deleteUserChunk((Chunk*)chunk);
 716     if (OK != rv) {
 717         systemDatabase_->releaseCheckpointMutex();
 718         printError(rv, "Unable to delete the chunk");
 719         return rv;
 720     }
 721     printDebug(DM_Database,"Deleted UserChunk:%x", chunk);
 722
 723     if (vcchunk != NULL) {
 724         rv = deleteUserChunk((Chunk*)vcchunk);
 725         if (OK != rv) {
 726             systemDatabase_->releaseCheckpointMutex();
 727             printError(rv, "Unable to delete the chunk");
 728             return rv;
 729         }
 730         printDebug(DM_Database,"Deleted UserChunk for Varchar:%x", chunk);
 731     }
 732
 733     //TODO::check whether indexes are available and drop that also.
 734     CatalogTableINDEX cIndex(systemDatabase_);
 735     int noIndexes = cIndex.getNumIndexes(tptr);
 736     for (int i =1 ; i<= noIndexes; i++) {
 737         char *idxName = cIndex.getIndexName(tptr, 1);
 738         dropIndexInt(idxName, false);
 739     }
 740     bool isFkExist=cFK.isFkTable(tptr);
 741     if(isFkExist)
 742     {
 743        dropForeignKey(tptr,false);
 744     }
 745     systemDatabase_->releaseCheckpointMutex();
 746     printDebug(DM_Database, "Deleted Table %s" , name);
 747     logFinest(Conf::logger, "Deleted Table %s" , name);
 748     rv = lMgr_->releaseLock(chunk);
 749     if (rv !=OK)
 750     {
 751         printError(ErrLockTimeOut, "Unable to release exclusive lock on the table\n");
 752         return rv;
 753     }
 754     return OK;
 755 }
 756
 757 //Return values: NULL for table not found
 758 Table* DatabaseManagerImpl::openTable(const char *name,bool checkpkfk)
 759 {
 760     DbRetVal ret = OK;
 761     //TODO::store table handles in list so that if it is
 762     //not closed by the application. destructor shall close it.
 763     TableImpl *table = new TableImpl();
 764     table->setDB(db_);
 765     table->setSystemDB(systemDatabase_);
 766     table->setLockManager(lMgr_);
 767     table->setTrans(ProcessManager::getThreadTransAddr(systemDatabase_->procSlot));
 768
 769     //to store the chunk pointer of table
 770     void *chunk = NULL;
 771     void *vcchunk = NULL;
 772
 773     //to store the tuple pointer of the table
 774     void *tptr =NULL;
 775
 776     //TODO::need to take shared lock on the table so that
 777     //all ddl operation will be denied on that table
 778     //which includes index creation, alter table
 779
 780     DbRetVal rv = systemDatabase_->getAllocDatabaseMutex();
 781     if (OK != rv) {
 782         printError(ErrSysInternal, "Unable to get database mutex");
 783         delete table;
 784         return NULL;
 785     }
 786     CatalogTableTABLE cTable(systemDatabase_);
 787     ret = cTable.getChunkAndTblPtr(name, chunk, tptr, vcchunk);
 788     if ( OK != ret)
 789     {
 790         systemDatabase_->releaseAllocDatabaseMutex();
 791         delete table;
 792         printError(ErrNotExists, "Table not exists %s", name);
 793         return NULL;
 794     }
 795     CTABLE *tTuple = (CTABLE*)tptr;
 796     table->setTableInfo(tTuple->tblName_, tTuple->tblID_, tTuple->length_,
 797                         tTuple->numFlds_, tTuple->numIndexes_,
 798                         tTuple->chunkPtr_, tTuple->varcharChunkPtr_);
 799     /*rv = table->lock(true); //take shared lock
 800     if (rv !=OK)
 801     {
 802         printError(ErrLockTimeOut, "Unable to acquire shared lock on the table\n");
 803         systemDatabase_->releaseAllocDatabaseMutex();
 804         delete table;
 805         return NULL;
 806     }*/
 807
 808
 809     if (tTuple->numFlds_ < 31)
 810     {
 811         table->isIntUsedForNULL = true;
 812         table->iNullInfo = 0;
 813         table->iNotNullInfo =0;
 814     }
 815     else
 816     {
 817         table->isIntUsedForNULL = false;
 818         int noFields = os::align(tTuple->numFlds_);
 819         table->cNullInfo = (char*) malloc(noFields);
 820         table->cNotNullInfo = (char*) malloc(noFields);
 821         for (int i =0 ; i < noFields; i++) table->cNullInfo[i] =0;
 822         for (int i =0 ; i < noFields; i++) table->cNotNullInfo[i] =0;
 823
 824     }
 825
 826     //get field information from FIELD table
 827     CatalogTableFIELD cField(systemDatabase_);
 828     table->ptrToAuto = cField.getFieldInfo(tptr, table->fldList_);
 829
 830     //populate the notnull info
 831     FieldIterator fIter = table->fldList_.getIterator();
 832     int fldpos=1;
 833     while (fIter.hasElement())
 834     {
 835         FieldDef *def = fIter.nextElement();
 836         if (table->isIntUsedForNULL) {
 837             if (def->isNull_) SETBIT(table->iNotNullInfo, fldpos);
 838         }
 839         else {
 840             if (def->isNull_) table->cNotNullInfo[fldpos-1] = 1;
 841         }
 842         fldpos++;
 843    }
 844
 845     //get the number of indexes on this table
 846     //and populate the indexPtr array
 847     CatalogTableINDEX cIndex(systemDatabase_);
 848     table->numIndexes_ = cIndex.getNumIndexes(tptr);
 849     if (table->numIndexes_) {
 850         table->indexPtr_ = new char*[table->numIndexes_];
 851         table->idxInfo = new IndexInfo*[table->numIndexes_];
 852     }
 853     else
 854     {
 855         table->indexPtr_ = NULL;
 856     }
 857     cIndex.getIndexPtrs(tptr, table->indexPtr_);
 858     for (int i =0 ; i < table->numIndexes_; i++ )
 859     {
 860         HashIndexInfo *hIdxInfo = new HashIndexInfo();
 861         CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 862         cIndexField.getFieldInfo(table->indexPtr_[i], hIdxInfo->idxFldList);
 863         ChunkIterator citer = CatalogTableINDEX::getIterator(table->indexPtr_[i]);
 864         hIdxInfo->indexPtr = table->indexPtr_[i];
 865         hIdxInfo->indType = ((CINDEX*)hIdxInfo->indexPtr)->indexType_;
 866         hIdxInfo->noOfBuckets = CatalogTableINDEX::getNoOfBuckets(table->indexPtr_[i]);
 867         FieldIterator fIter = hIdxInfo->idxFldList.getIterator();
 868         bool firstFld = true;
 869         while (fIter.hasElement())
 870         {
 871             FieldDef *def = fIter.nextElement();
 872             if (firstFld)
 873             {
 874                 hIdxInfo->fldOffset = table->fldList_.getFieldOffset(def->fldName_);
 875                 hIdxInfo->type = table->fldList_.getFieldType(def->fldName_);
 876                 hIdxInfo->compLength = table->fldList_.getFieldLength(def->fldName_);
 877                 firstFld = false;
 878             }else {
 879                 hIdxInfo->type = typeComposite;
 880                 hIdxInfo->compLength = hIdxInfo->compLength +
 881                          table->fldList_.getFieldLength(def->fldName_);
 882             }
 883         }
 884
 885         hIdxInfo->isUnique = CatalogTableINDEX::getUnique(table->indexPtr_[i]);
 886         hIdxInfo->buckets = (Bucket*)citer.nextElement();
 887         table->idxInfo[i] = (IndexInfo*) hIdxInfo;
 888     }
 889     systemDatabase_->releaseAllocDatabaseMutex();
 890     //Foreign key Operation
 891     if(checkpkfk){
 892         CatalogTableFK cFk(systemDatabase_);
 893         int totalFld=0;
 894         table->numFkRelation_ = cFk.getNumFkTable(tptr);
 895         if (table->numFkRelation_) {
 896             table->isPkTbl=true;//TODO:for Delete In casecade
 897             totalFld=cFk.getNoOfFkTable(tptr);
 898             //printDebug(DM_TEST,"Total table is %d\n",totalFld);
 899             char **fptr = new char* [totalFld];
 900             cFk.getFkTableName(tptr,fptr);
 901             for(int count=0; count < totalFld; count++){
 902                //printDebug(DM_TEST,"FK Name is %s\n",fptr[count]);
 903                Table *pkTable=openTable(fptr[count],false);
 904                if (pkTable) table->tblFkList.append(pkTable);
 905                else {
 906                    printError(ErrSysInternal, "Unable to open foreign key tables");
 907                    delete[] fptr;
 908                    pkTable->close();
 909                    return NULL;
 910                }
 911             }
 912             delete[] fptr;
 913
 914         }
 915         char *tblName = NULL;
 916         table->isFkTbl = cFk.isFkTable(tptr);
 917         if(table->isFkTbl)
 918         {
 919             totalFld=cFk.getNoOfPkTable(tptr);
 920             char **fptr = new char* [totalFld];
 921             cFk.getPkTableName(tptr,fptr);
 922             for(int count=0; count<totalFld; count++){
 923                 //printDebug(DM_TEST,"Parent Name is %s\n",fptr[count]);
 924                 Table *fkTable = openTable(fptr[count],false);
 925                 if (fkTable) table->tblList.append(fkTable);
 926                 else {
 927                    printError(ErrSysInternal, "Unable to open foreign key tables");
 928                    delete[] fptr;
 929                    fkTable->close();
 930                    return NULL;
 931                }
 932             }
 933             delete[] fptr;
 934         }
 935     }
 936     printDebug(DM_Database,"Opening table handle name:%s chunk:%x numIndex:%d",
 937                                          name, chunk, table->numIndexes_);
 938     logFinest(Conf::logger, "Opening Table %s" , name);
 939     return table;
 940 }
 941
 942 List DatabaseManagerImpl::getAllTableNames(int *retval)
 943 {
 944     DbRetVal ret = OK;
 945     //to store the tuple pointer of the table
 946     void *tptr =NULL;
 947
 948     /*DbRetVal rv = systemDatabase_->getSCheckpointMutex();
 949     if (OK != rv) {
 950         printError(ErrSysInternal, "Unable to get checkpoint mutex");
 951         if(retval) *retval = rv;
 952         List tableList;
 953         return tableList;
 954     }*/
 955     CatalogTableTABLE cTable(systemDatabase_);
 956     List tableList = cTable.getTableList();
 957     //systemDatabase_->releaseCheckpointMutex();
 958     return tableList;
 959 }
 960
 961
 962
 963
 964 //Return values: -1 for table not found
 965 void DatabaseManagerImpl::closeTable(Table *table)
 966 {
 967     printDebug(DM_Database,"Closing table handle: %x", table);
 968     if (NULL == table) return;
 969     //table->unlock();
 970  /*   TableImpl *fkTbl =NULL;
 971     ListIterator tblIter = ((TableImpl*)table)->tblList.getIterator();
 972     tblIter.reset();
 973     while (tblIter.hasElement()){
 974         fkTbl = (TableImpl *) tblIter.nextElement();
 975         closeTable(fkTbl);
 976     }
 977     ((TableImpl*)table)->tblList.reset();
 978     tblIter = ((TableImpl*)table)->tblFkList.getIterator();
 979     tblIter.reset();
 980     while (tblIter.hasElement()){
 981         fkTbl = (TableImpl *) tblIter.nextElement();
 982         closeTable(fkTbl);
 983     }
 984     ((TableImpl*)table)->tblFkList.reset();*/
 985     if (table) delete table; table = NULL;
 986     logFinest(Conf::logger, "Closing Table");
 987 }
 988
 989 DbRetVal DatabaseManagerImpl::createIndex(const char *indName, IndexInitInfo *info)
 990 {
 991     DbRetVal rv = OK;
 992     if (!info->isUnique && info->isPrimary)
 993     {
 994         printError(ErrBadCall, "Primary key cannot be non unique\n");
 995         return ErrBadCall;
 996     }
 997     if (!Util::isIdentifier((char*)indName)) {
 998         printError(ErrBadArg, "Invalid character for index name");
 999         return ErrBadArg;
1000     }
1001
1002     if (info->indType == hashIndex)
1003     {
1004         //Assumes info is of type HashIndexInitInfo
1005         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
1006         rv = createHashIndex(indName, info->tableName, info->list, hInfo->bucketSize,
1007                              info->isUnique, info->isPrimary);
1008     }
1009     else if (info->indType == treeIndex)
1010     {
1011         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
1012         rv = createTreeIndex(indName, info->tableName, info->list,
1013                           hInfo->bucketSize, info->isUnique, info->isPrimary);
1014
1015     }else {
1016         printError(ErrBadCall, "Index type not supported\n");
1017         return ErrBadCall;
1018     }
1019     return rv;
1020 }
1021
1022
1023 //-1 -> Table does not exists
1024 //-2 -> Field does not exists
1025 //-3 -> bucketSize is not valid
1026 DbRetVal DatabaseManagerImpl::createHashIndex(const char *indName, const char *tblName,
1027                       FieldNameList &fldList, int bucketSize, bool isUnique, bool isPrimary)
1028 {
1029     //validate the bucket size
1030     if (bucketSize < 100 || bucketSize > 200000)
1031     {
1032         printError(ErrBadRange, "Index Bucket size %d not in range 100-200000",
1033                                  bucketSize);
1034         return ErrBadRange;
1035     }
1036     int totFlds = fldList.size();
1037     if (totFlds == 0)
1038     {
1039         printError(ErrBadCall, "No Field name specified");
1040         return ErrBadCall;
1041     }
1042     void *tptr =NULL;
1043     void *chunk = NULL;
1044     void *vcchunk = NULL;
1045     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
1046     if (OK != rv)
1047     {
1048         printError(ErrSysInternal, "Unable to get database mutex");
1049         return ErrSysInternal;
1050     }
1051
1052     //check whether table exists
1053     CatalogTableTABLE cTable(systemDatabase_);
1054     cTable.getChunkAndTblPtr(tblName, chunk, tptr, vcchunk);
1055     if (NULL == tptr)
1056     {
1057         systemDatabase_->releaseCheckpointMutex();
1058         printError(ErrNotExists, "Table does not exist %s", tblName);
1059         return ErrNotExists;
1060     }
1061
1062     //check whether field exists
1063     char **fptr = new char* [totFlds];
1064     CatalogTableFIELD cField(systemDatabase_);
1065     rv = cField.getFieldPtrs(fldList, tptr, fptr);
1066     if (OK != rv)
1067     {
1068         delete[] fptr;
1069         systemDatabase_->releaseCheckpointMutex();
1070         //TODO::check test cases of dbapi/Index, they give wrong results
1071         //if (rv == ErrBadCall) {
1072         ////    if (isPrimary) printError(ErrBadCall, "Field can have NULL values");
1073         //} else {
1074         //printError(ErrNotExists, "Field does not exist");
1075         //}
1076         //return ErrBadCall;
1077         if (rv != ErrBadCall) {
1078             printError(ErrNotExists, "Field does not exist");
1079             return ErrNotExists;
1080         }
1081     }
1082     for (int i=0; i <totFlds; i++)
1083     {
1084         CFIELD* fInfo = (CFIELD*)fptr[i];
1085         if (fInfo->type_ == typeFloat || fInfo->type_ == typeDouble || fInfo->type_ == typeTimeStamp)
1086         {
1087             printError(ErrBadArg, "HashIndex cannot be created for float or double or timestamp type");
1088             delete[] fptr;
1089             systemDatabase_->releaseCheckpointMutex();
1090             return ErrBadArg;
1091         }
1092         if (!fInfo->isNull_ && isPrimary )
1093         {
1094             printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
1095             delete[] fptr;
1096             systemDatabase_->releaseCheckpointMutex();
1097             return ErrBadArg;
1098         }
1099          if(isPrimary){fInfo->isPrimary_=true;fInfo->isUnique_=true;}
1100          if(isUnique){fInfo->isUnique_=true;}
1101     }
1102     //create chunk to store the meta data of the index created
1103     //for latches and bucket pointers
1104     printDebug(DM_HashIndex, "Creating chunk for storing hash buckets of size %d\n",
1105                              bucketSize * sizeof(Bucket));
1106     Chunk* chunkInfo = createUserChunk(bucketSize * sizeof(Bucket));
1107     if (NULL == chunkInfo)
1108     {
1109         delete[] fptr;
1110         systemDatabase_->releaseCheckpointMutex();
1111         printError(ErrSysInternal, "Unable to create chunk");
1112         return ErrSysInternal;
1113     }
1114     chunkInfo->setChunkName(indName);
1115     //create memory for holding the bucket pointers
1116     void *buckets = chunkInfo->allocate(db_, &rv);
1117     if (NULL == buckets)
1118     {
1119         delete[] fptr;
1120         deleteUserChunk(chunkInfo);
1121         systemDatabase_->releaseCheckpointMutex();
1122         printError(rv, "Unable to allocate memory for bucket");
1123         return rv;
1124     }
1125     Bucket *buck = (Bucket*) buckets;
1126     initHashBuckets(buck, bucketSize);
1127
1128     //create chunk to store the hash index nodes
1129     Chunk* hChunk = createUserChunk(sizeof(HashIndexNode));
1130     if (NULL == hChunk)
1131     {
1132         delete[] fptr;
1133         deleteUserChunk(chunkInfo);
1134         systemDatabase_->releaseCheckpointMutex();
1135         printError(ErrSysInternal, "Unable to create chunk for storing hash index nodes");
1136         return ErrSysInternal;
1137     }
1138
1139      hChunk->setChunkName(indName);
1140     //add row to INDEX
1141     void *tupleptr = NULL;
1142     CatalogTableINDEX cIndex(systemDatabase_);
1143     rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
1144                         chunkInfo, bucketSize, hChunk, tupleptr);
1145     if (OK != rv)
1146     {
1147         delete[] fptr;
1148         deleteUserChunk(hChunk);
1149         deleteUserChunk(chunkInfo);
1150         systemDatabase_->releaseCheckpointMutex();
1151         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
1152         return ErrSysInternal;
1153     }
1154     //add rows to INDEXFIELD
1155     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
1156     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
1157
1158     if (OK != rv)
1159     {
1160         delete[] fptr;
1161         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
1162         deleteUserChunk(hChunk);
1163         deleteUserChunk(chunkInfo);
1164         systemDatabase_->releaseCheckpointMutex();
1165         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
1166         return ErrSysInternal;
1167     }
1168     delete[] fptr;
1169     systemDatabase_->releaseCheckpointMutex();
1170
1171     //TODO:: Take table lock
1172
1173     // Following code is written by Kishor Amballi
1174     TableImpl *tbl = (TableImpl *) openTable(tblName);
1175     if (NULL == tbl) {
1176         printError(ErrSysInternal, "Unable to open table %s", tblName);
1177         return ErrSysInternal;
1178     }
1179     if (! tbl->numTuples()) {
1180         printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x", indName, tblName, buckets);
1181         logFinest(Conf::logger, "Creating HashIndex %s on %s with bucket size %d", indName, tblName, buckets);
1182         closeTable(tbl);
1183         return OK;
1184     }
1185     HashIndexInfo *indxInfo = NULL;
1186     int i = 0;
1187     for (i = 0; i < tbl->numIndexes_; i++) {
1188         if(((HashIndexInfo *)tbl->idxInfo[i])->indexPtr == tupleptr) {
1189             indxInfo = (HashIndexInfo *) tbl->idxInfo[i];
1190             break;
1191         }
1192     }
1193     void *recPtr = NULL;
1194     ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
1195     tbl->setLoading(true);
1196     while ((recPtr = chIter.nextElement()) != NULL) {
1197         rv = tbl->insertIndexNode(*tbl->trans, tupleptr, indxInfo, recPtr);
1198         if (rv == ErrUnique) {
1199             closeTable(tbl);
1200             dropIndex(indName);
1201             return rv;
1202         }
1203     }
1204     closeTable(tbl);
1205     printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x", indName, tblName, buckets);
1206     logFinest(Conf::logger, "Creating HashIndex %s on %s with bucket size %d", indName, tblName, buckets);
1207     return OK;
1208 }
1209
1210
1211 DbRetVal DatabaseManagerImpl::createTreeIndex(const char *indName, const char *tblName,
1212                       FieldNameList &fldList, int nodeSize, bool isUnique, bool isPrimary)
1213 {
1214     if (nodeSize < 20 || nodeSize > 20000)
1215     {
1216         printError(ErrBadRange,"Tree Index Node size %d not in range 20-20000",
1217                                  nodeSize);
1218         return ErrBadRange;
1219     }
1220     int totFlds = fldList.size();
1221     if (totFlds == 0) {
1222         printError(ErrBadCall, "No Field name specified");
1223         return ErrBadCall;
1224     }else if (totFlds != 1) {
1225         printError(ErrBadCall, "Composite index not supported for Tree");
1226         return ErrBadCall;
1227     }
1228     void *tptr =NULL;
1229     void *chunk = NULL;
1230     void *vcchunk = NULL;
1231     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
1232     if (OK != rv)
1233     {
1234         printError(ErrSysInternal, "Unable to get database mutex");
1235         return ErrSysInternal;
1236     }
1237     //check whether table exists
1238
1239     CatalogTableTABLE cTable(systemDatabase_);
1240     cTable.getChunkAndTblPtr(tblName, chunk, tptr, vcchunk);
1241     if (NULL == tptr)
1242     {
1243         systemDatabase_->releaseCheckpointMutex();
1244         printError(ErrNotExists, "Table does not exist %s", tblName);
1245         return ErrNotExists;
1246     }
1247     char **fptr = new char* [totFlds];
1248     CatalogTableFIELD cField(systemDatabase_);
1249     rv = cField.getFieldPtrs(fldList, tptr, fptr);
1250     if (OK != rv)
1251     {
1252         delete[] fptr;
1253         systemDatabase_->releaseCheckpointMutex();
1254         if (rv != ErrBadCall) {
1255             printError(ErrNotExists, "Field does not exist");
1256             return ErrNotExists;
1257         }
1258     }
1259     for (int i=0; i <totFlds; i++)
1260     {
1261         CFIELD* fInfo = (CFIELD*)fptr[i];
1262         if (!fInfo->isNull_ && isPrimary )
1263         {
1264             printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
1265             delete[] fptr;
1266             systemDatabase_->releaseCheckpointMutex();
1267             return ErrBadArg;
1268         }
1269     }
1270     int chunkSize = sizeof(TreeNode)+(nodeSize * sizeof(void*));
1271     printDebug(DM_HashIndex, "Creating chunk for storing tree nodes of size %d\n", chunkSize);
1272
1273     Chunk* chunkInfo = createUserChunk(chunkSize);
1274     if (NULL == chunkInfo)
1275     {
1276         delete[] fptr;
1277         systemDatabase_->releaseCheckpointMutex();
1278         printError(ErrSysInternal, "Unable to create chunk");
1279         return ErrSysInternal;
1280     }
1281
1282
1283     void *tupleptr = NULL;
1284
1285     CatalogTableINDEX cIndex(systemDatabase_);
1286     rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
1287                         chunkInfo, nodeSize, NULL, tupleptr);
1288     if (OK != rv)
1289     {
1290         delete[] fptr;
1291         deleteUserChunk(chunkInfo);
1292         systemDatabase_->releaseCheckpointMutex();
1293         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
1294         return ErrSysInternal;
1295     }
1296     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
1297     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
1298
1299     if (OK != rv)
1300     {
1301         delete[] fptr;
1302         void *hChunk = NULL;
1303         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
1304         deleteUserChunk(chunkInfo);
1305         systemDatabase_->releaseCheckpointMutex();
1306         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
1307         return ErrSysInternal;
1308     }
1309     delete[] fptr;
1310     systemDatabase_->releaseCheckpointMutex();
1311     //TODO::if tuples already present in this table, then create tree index '
1312     //nodes
1313     TableImpl *tbl = (TableImpl *) openTable(tblName);
1314     if (NULL == tbl) {
1315         printError(ErrSysInternal, "Unable to open table %s", tblName);
1316         return ErrSysInternal;
1317     }
1318     if (! tbl->numTuples()) {
1319         printDebug(DM_Database, "Creating Tree Index Name:%s tblname:%s node size:%x",indName, tblName, nodeSize);
1320         logFinest(Conf::logger, "Creating TreeIndex %s on %s with node size %d",indName, tblName, nodeSize);
1321         closeTable(tbl);
1322         return OK;
1323     }
1324     HashIndexInfo *indxInfo = NULL;
1325     int i = 0;
1326     for (i = 0; i < tbl->numIndexes_; i++) {
1327         if(((HashIndexInfo *)tbl->idxInfo[i])->indexPtr == tupleptr) {
1328             indxInfo = (HashIndexInfo *) tbl->idxInfo[i];
1329             break;
1330         }
1331     }
1332     void *recPtr = NULL;
1333     ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
1334     tbl->setLoading(true);
1335     while ((recPtr = chIter.nextElement()) != NULL) {
1336         rv = tbl->insertIndexNode(*tbl->trans, tupleptr, indxInfo, recPtr);
1337         if (rv == ErrUnique) {
1338             dropIndex(indName);
1339             closeTable(tbl);
1340             return rv;
1341         }
1342     }
1343     closeTable(tbl);
1344     printDebug(DM_Database, "Creating Tree Index Name:%s tblname:%s node size:%x",
1345                                    indName, tblName, nodeSize);
1346     logFinest(Conf::logger, "Creating TreeIndex %s on %s with node size %d",
1347                                    indName, tblName, nodeSize);
1348     return OK;
1349 }
1350
1351
1352
1353 void DatabaseManagerImpl::initHashBuckets(Bucket *buck, int bucketSize)
1354 {
1355     os::memset((void*)buck, 0, bucketSize * sizeof(Bucket));
1356
1357     for (int i=0; i < bucketSize ; i++)
1358     {
1359         buck[i].mutex_.init("Bucket");
1360     }
1361     return;
1362 }
1363
1364 DbRetVal DatabaseManagerImpl::dropIndex(const char *name)
1365 {
1366     return dropIndexInt(name, true);
1367 }
1368
1369 DbRetVal DatabaseManagerImpl::dropIndexInt(const char *name, bool takeLock)
1370 {
1371     DbRetVal rv = OK;
1372     void *chunk = NULL, *hchunk = NULL;
1373     void *tptr =NULL;
1374     int ret = 0;
1375     if (takeLock) {
1376         rv = systemDatabase_->getXCheckpointMutex();
1377         if (OK != rv)
1378         {
1379             printError(ErrSysInternal, "Unable to get database mutex");
1380             return ErrSysInternal;
1381         }
1382     }
1383
1384     //remove the entry in INDEX
1385     CatalogTableINDEX cIndex(systemDatabase_);
1386     rv = cIndex.remove(name, chunk, hchunk, tptr);
1387     if (OK != rv)
1388     {
1389         if (takeLock) systemDatabase_->releaseCheckpointMutex();
1390         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
1391         return ErrSysInternal;
1392     }
1393     printDebug(DM_Database, "Removing from INDEX %s",name);
1394     //remove the entries in the INDEXFIELD table
1395     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
1396     rv = cIndexField.remove(tptr);
1397     if (OK != rv)
1398     {
1399         if (takeLock) systemDatabase_->releaseCheckpointMutex();
1400         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
1401         return ErrSysInternal;
1402     }
1403     printDebug(DM_Database, "Removing from INDEXFIELD %s",name);
1404
1405     //delete the index chunk
1406     CINDEX *iptr = (CINDEX*)tptr;
1407     rv = deleteUserChunk((Chunk*)chunk);
1408     if (OK != rv)
1409     {
1410         if (takeLock) systemDatabase_->releaseCheckpointMutex();
1411         printError(ErrSysInternal, "Unable to delete the index chunk");
1412         return ErrSysInternal;
1413     }
1414     //delete the index hash node chunk
1415     if (iptr->indexType_ == hashIndex) {
1416         rv = deleteUserChunk((Chunk*)hchunk);
1417         if (OK != rv)
1418         {
1419             if (takeLock) systemDatabase_->releaseCheckpointMutex();
1420             printError(ErrSysInternal, "Unable to delete the index hash node chunk");
1421             return ErrSysInternal;
1422         }
1423     }
1424     if (takeLock) systemDatabase_->releaseCheckpointMutex();
1425
1426     //TODO::If tuples present in this table, then
1427     //free all hash index nodes for this table.
1428     //free all nodes in list of all buckets
1429     //Take table lock
1430
1431     printDebug(DM_Database, "Dropped hash index %s",name);
1432     logFinest(Conf::logger, "Deleted Index %s", name);
1433     return OK;
1434 }
1435 DbRetVal DatabaseManagerImpl::createForeignKey(char *fKName,ForeignKeyInfo *info)
1436 {
1437    DbRetVal rv = OK;
1438    int totFkFlds = info->fkFldList.size();
1439    int totPkFlds = info->pkFldList.size();
1440    if (totFkFlds==0 && totPkFlds==0) {
1441         printError(ErrBadCall, "No Field name specified");
1442         return ErrBadCall;
1443    }
1444     void *tptr =NULL;
1445     void *chunk = NULL;
1446     void *vcchunk = NULL;
1447     rv = systemDatabase_->getXCheckpointMutex();
1448     if (OK != rv)
1449     {
1450         printError(ErrSysInternal, "Unable to get database mutex");
1451         return ErrSysInternal;
1452     }
1453     CatalogTableTABLE cTable(systemDatabase_);
1454     cTable.getChunkAndTblPtr(info->fkTableName, chunk, tptr, vcchunk);
1455     if (NULL == tptr)
1456     {
1457         systemDatabase_->releaseCheckpointMutex();
1458         printError(ErrNotExists, "Table does not exist %s", info->fkTableName);
1459         return ErrNotExists;
1460     }
1461     char **fptr = new char* [totFkFlds];
1462     CatalogTableFIELD cField(systemDatabase_);
1463     rv = cField.getFieldPtrs(info->fkFldList, tptr, fptr);
1464     if (OK != rv)
1465     {
1466         delete[] fptr;
1467         systemDatabase_->releaseCheckpointMutex();
1468         if (rv != ErrBadCall) {
1469             printError(ErrNotExists, "Field does not exist");
1470             return ErrNotExists;
1471         }
1472     }
1473     void *tPkptr =NULL;
1474     void *chunkPk = NULL;
1475     void *vcchunkPk = NULL;
1476     CatalogTableTABLE c2Table(systemDatabase_);
1477     c2Table.getChunkAndTblPtr(info->pkTableName, chunkPk, tPkptr, vcchunkPk);
1478     if (NULL == tPkptr)
1479     {
1480         systemDatabase_->releaseCheckpointMutex();
1481         printError(ErrNotExists, "Table does not exist %s", info->pkTableName);
1482         return ErrNotExists;
1483     }
1484     char **fPkptr = new char* [totPkFlds];
1485     CatalogTableFIELD c2Field(systemDatabase_);
1486     rv = c2Field.getFieldPtrs(info->pkFldList, tPkptr, fPkptr);
1487     if (OK != rv)
1488     {
1489         delete[] fptr;
1490         delete[] fPkptr;
1491         systemDatabase_->releaseCheckpointMutex();
1492         if (rv != ErrBadCall) {
1493             printError(ErrNotExists, "Field does not exist");
1494             return ErrNotExists;
1495         }
1496     }
1497     //Create New chunkdatanode
1498     CatalogTableFK cFK(systemDatabase_);
1499     rv = cFK.insert(fKName, tptr, tPkptr);//TODO
1500     if (OK != rv)
1501     {
1502         delete[] fptr;
1503         delete[] fPkptr;
1504         systemDatabase_->releaseCheckpointMutex();
1505         printError(ErrSysInternal, "Catalog table updation failed in CFK table");
1506         return ErrSysInternal;
1507     }
1508
1509     CatalogTableFKFIELD cFKField(systemDatabase_);
1510     rv = cFKField.insert(fKName,fptr,fPkptr,totFkFlds);
1511     if (OK != rv)
1512     {
1513         delete[] fptr;
1514         delete[] fPkptr;
1515         cFK.remove(tptr);
1516         systemDatabase_->releaseCheckpointMutex();
1517         printError(ErrSysInternal, "Catalog table updation failed in CFKFIELD table");
1518         return ErrSysInternal;
1519     }
1520     systemDatabase_->releaseCheckpointMutex();
1521     delete[] fptr;
1522     delete[] fPkptr;
1523     return rv;
1524 }
1525 DbRetVal DatabaseManagerImpl::dropForeignKey(void *tptr,bool trylock)
1526 {
1527     DbRetVal rv = OK;
1528     if(trylock){
1529         rv = systemDatabase_->getXCheckpointMutex();
1530         if (OK != rv)
1531         {
1532             printError(ErrSysInternal, "Unable to get database mutex");
1533             return ErrSysInternal;
1534         }
1535     }
1536     void *fkChunk=NULL;
1537     CatalogTableFK cFK(systemDatabase_);
1538     int total = cFK.getNoOfPkTable(tptr);
1539     //printDebug(DM_TEST,"total fk chunk %d",total);
1540     for (int i=0;i< total; i++)
1541     {
1542          fkChunk = cFK.getFkCTable(tptr);
1543          if(NULL==fkChunk)
1544          {
1545             if(trylock){
1546                 systemDatabase_->releaseCheckpointMutex();
1547             }
1548             printError(ErrSysInternal, "Catalog table not finds CFKFIELD table");
1549             return ErrSysInternal;
1550          }
1551          CatalogTableFKFIELD cFKField(systemDatabase_);
1552          rv = cFKField.remove(fkChunk);
1553          if (OK != rv)
1554          {
1555              if(trylock){
1556                 systemDatabase_->releaseCheckpointMutex();
1557              }
1558              printError(ErrSysInternal, "Catalog table updation failed in CFKFIELD table");
1559              return ErrSysInternal;
1560          }
1561          rv =cFK.remove(fkChunk);
1562          if (OK != rv)
1563          {
1564             if(trylock){
1565                 systemDatabase_->releaseCheckpointMutex();
1566             }
1567             printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
1568             return ErrSysInternal;
1569          }
1570     }
1571     if(trylock){
1572         systemDatabase_->releaseCheckpointMutex();
1573     }
1574     return rv;
1575 }
1576
1577 void DatabaseManagerImpl::printTreeIndexNodeInfo(char *name, bool flag)
1578 {
1579     CatalogTableINDEX cIndex(systemDatabase_);
1580     DbRetVal rv = OK;
1581     void *chunk = NULL, *hchunk = NULL;
1582     void *tptr =NULL;
1583     rv = cIndex.get(name, chunk, hchunk, tptr);
1584     if (OK != rv) return;
1585     IndexType iType = CatalogTableINDEX::getType(tptr);
1586     if (treeIndex != iType)
1587     {
1588        printf("%s is not a tree index\n ");
1589        return;
1590     }
1591     Chunk *ch = (Chunk*) chunk;
1592     if(flag){  if(hchunk)((TreeNode*) hchunk)->displayAll(); }
1593     else {
1594         int offset = CatalogTableINDEX::getOffsetOfFirstField(tptr);
1595         //if(typeInt != offset) { printf("%s is not on Integer Type Field. To see info Index should be on integer type field. \n "); return;}
1596         if(hchunk) ((TreeNode*) hchunk)->displayAll(offset);
1597     }
1598 }
1599
1600 DbRetVal DatabaseManagerImpl::printIndexInfo(char *name)
1601 {
1602     CatalogTableINDEX cIndex(systemDatabase_);
1603     DbRetVal rv = OK;
1604     void *chunk = NULL, *hchunk = NULL;
1605     void *tptr =NULL;
1606     rv = cIndex.get(name, chunk, hchunk, tptr);
1607     if (OK != rv) return rv;
1608     printf("<IndexName> %s </IndexName>\n", name);
1609     printf("<Unique> %d </Unique>\n", CatalogTableINDEX::getUnique(tptr));
1610     IndexType iType = CatalogTableINDEX::getType(tptr);
1611     if(hashIndex == iType)
1612         printf("<Type> Hash Index </Type>\n");
1613     else if (treeIndex == iType)
1614         printf("<Type> Tree Index </Type>\n");
1615     else
1616         printf("<Type> Unknown Index </Type>\n");
1617
1618     Chunk *ch = (Chunk*) chunk;
1619     printf("<HashBucket>\n");
1620     printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
1621     printf("  <TotalBuckets> %d </TotalBuckets> \n", CatalogTableINDEX::getNoOfBuckets(tptr));
1622     printf("</HashBucket>\n");
1623     if(treeIndex != iType){
1624         ch = (Chunk*) hchunk;
1625         printf("<IndexNodes>\n");
1626         printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
1627         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
1628         printf("<IndexNodes>\n");
1629     }else{
1630         printf("<IndexNodes>\n");
1631         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
1632         if(hchunk)
1633             printf("  <TotalElements> %lld </TotalElements>\n",((TreeNode*) hchunk)->getTotalElements());
1634         else
1635             printf("  <TotalElements> 0 </TotalElements>\n");
1636         printf("<IndexNodes>\n");
1637     }
1638     return OK;
1639 }
1640
1641 DbRetVal DatabaseManagerImpl::registerThread()
1642 {
1643     DbRetVal rv = OK;
1644     if (pMgr_ != NULL)
1645     {
1646         printError(ErrAlready, "Process already registered\n");
1647         return ErrAlready;
1648     }
1649     pMgr_ = new ProcessManager();
1650     rv = pMgr_->registerThread();
1651     if (rv ==OK) { procSlot = pMgr_->getProcSlot();
1652         printDebug(DM_Process, "Process registed with slot %d\n", procSlot);
1653     }
1654     return rv;
1655 }
1656
1657 DbRetVal DatabaseManagerImpl::deregisterThread()
1658 {
1659     DbRetVal rv = OK;
1660     if (pMgr_ != NULL)
1661     {
1662         rv = pMgr_->deregisterThread(procSlot);
1663         delete pMgr_;
1664         pMgr_ = NULL;
1665     }
1666     return rv;
1667 }
1668
1669 bool DatabaseManagerImpl::isAnyOneRegistered()
1670 {
1671     if (pMgr_ != NULL) return pMgr_->isAnyOneRegistered();
1672     return true;
1673 }
1674
1675
1676 void DatabaseManagerImpl::printUsageStatistics()
1677 {
1678     pMgr_->printUsageStatistics();
1679     tMgr_->printUsageStatistics();
1680     lMgr_->printUsageStatistics();
1681 }
1682
1683 void DatabaseManagerImpl::printDebugLockInfo()
1684 {
1685     lMgr_->printDebugInfo();
1686 }
1687
1688 void DatabaseManagerImpl::printDebugTransInfo()
1689 {
1690     tMgr_->printDebugInfo(systemDatabase_);
1691 }
1692 void DatabaseManagerImpl::printDebugProcInfo()
1693 {
1694     pMgr_->printDebugInfo();
1695 }
1696 void DatabaseManagerImpl::printDebugChunkInfo()
1697 {
1698     printf("<NotYetImplemented> </NotYetImplemented>\n");
1699 }
1700 ChunkIterator DatabaseManagerImpl::getSystemTableIterator(CatalogTableID id)
1701 {
1702     Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(id);
1703     return fChunk->getIterator();
1704 }
1705
1706 Chunk* DatabaseManagerImpl::getSystemTableChunk(CatalogTableID id)
1707 {
1708     return systemDatabase_->getSystemDatabaseChunk(id);
1709 }
1710
1711 int DatabaseManagerImpl::getNoOfPagesForTable(char *tblName)
1712 {
1713     Table *tbl = openTable(tblName);
1714     if (NULL == tbl) {
1715         printError(ErrSysInternal, "Unable to open table %s", tblName);
1716         return 0;
1717     }
1718     TableImpl *tb = (TableImpl *) tbl;
1719     int pages = 0;
1720     if (tb->numTuples()) pages = tb->pagesUsed();
1721     closeTable(tbl);
1722     return pages;
1723 }
1724
1725 DbRetVal DatabaseManagerImpl::loadRecords(char *tblName, char *buffer)
1726 {
1727     // buffer should be as big as the no of pages occupied by the records
1728     Table *tbl = openTable(tblName);
1729     if (NULL == tbl) {
1730         printError(ErrSysInternal, "Unable to open table %s", tblName);
1731         return ErrSysInternal;
1732     }
1733     TableImpl *tb = (TableImpl *) tbl;
1734     char *bufIter = buffer;
1735     int pages = *(int *) bufIter; bufIter += sizeof(int);
1736     Page *firstPage = ((Chunk *)(tb->chunkPtr_))->getFirstPage();
1737     PageInfo *pi = (PageInfo *) firstPage;
1738     memcpy(bufIter, pi, PAGE_SIZE);
1739     bufIter += PAGE_SIZE;
1740     for (int i = 0; i < pages - 1; i++) {
1741         Page *nPage = pi->nextPage_;
1742         memcpy(bufIter, nPage, PAGE_SIZE);
1743         bufIter += PAGE_SIZE;
1744         pi = (PageInfo *) nPage;
1745     }
1746     closeTable(tbl);
1747     return OK;
1748 }
1749
1750 DbRetVal DatabaseManagerImpl::pasteRecords(char *tblName, void *buffer)
1751 {
1752     // buffer should be as big as the no of pages occupied by the records
1753     Table *tbl = openTable(tblName);
1754     if (NULL == tbl) {
1755         printError(ErrSysInternal, "Unable to open table %s", tblName);
1756         return ErrSysInternal;
1757     }
1758     TableImpl *tb = (TableImpl *) tbl;
1759     Database *db = tb->getDB();
1760     char *bufIter = (char *) buffer;
1761     int pages = *(int *) bufIter;
1762     bufIter += sizeof(int);
1763
1764     Page *firstPage = ((Chunk *)(tb->chunkPtr_))->getFirstPage();
1765     PageInfo *pi = (PageInfo *) firstPage;
1766     memcpy(pi, bufIter, PAGE_SIZE);
1767     bufIter += PAGE_SIZE;
1768     while (--pages != 0) {
1769         //get a new page allocated
1770         Page *newPage = db->getFreePage();
1771         memcpy(newPage, bufIter, PAGE_SIZE);
1772         pi->nextPage_ = newPage;
1773         pi = (PageInfo *) newPage;
1774     }
1775     // initialize chunk details and pageInfo
1776     ((Chunk *)tb->chunkPtr_)->curPage_ = pi;
1777     closeTable(tbl);
1778     return OK;
1779 }
1780
1781 DbRetVal DatabaseManagerImpl::checkPoint()
1782 {
1783     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
1784     if (OK != rv ) {
1785         printError(rv, "Unable to get checkpoint mutex");
1786         return ErrLockTimeOut;
1787     }
1788     if (tMgr_ && !tMgr_->isTransactionConsistent(systemDatabase_)) {
1789         printf("not in transaction consistent point\n");
1790         systemDatabase_->releaseCheckpointMutex();
1791         return ErrLockTimeOut;
1792     }
1793     rv = writeSchemaFile();
1794     if (rv != OK) {
1795         printError(ErrSysInternal, "checkpoint error");
1796     }
1797     rv = db()->checkPoint();
1798     systemDatabase_->releaseCheckpointMutex();
1799     return rv;
1800 }
1801
1802 DbRetVal DatabaseManagerImpl::writeSchemaFile()
1803 {
1804     DbRetVal rv = OK;
1805     FILE *fp = NULL;
1806     FILE *fp1 = NULL;
1807     int fd = -1;
1808     char schFile[MAX_FILE_LEN];
1809     char mapFile[MAX_FILE_LEN];
1810     sprintf(schFile, "%s/db.chkpt.schema1", Conf::config.getDbFile());
1811     sprintf(mapFile, "%s/db.chkpt.map1", Conf::config.getDbFile());
1812     fp = fopen(schFile, "r");
1813     if (fp != NULL) {
1814         fclose(fp);
1815         int ret = unlink(schFile);
1816         if( ret != 0) {
1817             printError(ErrOS, "checkpoint: delete schema file failed");
1818             return ErrOS;
1819         }
1820     }
1821     fp = fopen(schFile, "w+");
1822     if (fp == NULL) {
1823             printError(ErrOS, "Unable to create schema file for chkpt.");
1824             return ErrOS;
1825     }
1826     fp1 = fopen(mapFile, "r");
1827     if (fp1 != NULL) {
1828         fclose(fp1);
1829         int ret = unlink(mapFile);
1830         if( ret != 0) {
1831             printError(ErrOS, "checkpoint: delete schema file failed");
1832             return ErrOS;
1833         }
1834     }
1835     fd = open(mapFile, O_WRONLY|O_CREAT, 0644);
1836     if (fd == -1) {
1837         printError(ErrOS, "checkpoint: Unable to create map file.");
1838         return ErrOS;
1839     }
1840     List tableList = getAllTableNames();
1841     ListIterator iter = tableList.getIterator();
1842     Identifier *elem = NULL;
1843     int count =0;
1844     while (iter.hasElement()) {
1845         elem = (Identifier*) iter.nextElement();
1846 //        if (TableConf::config.isTableCached(elem->name) == OK) continue;
1847         fprintf(fp, "CREATE TABLE %s (", elem->name);
1848         Table *table = openTable(elem->name);
1849         if (NULL == table) {
1850             printError(ErrSysInternal, "Unable to open table %s", elem->name);
1851             return ErrSysInternal;
1852         }
1853         void *chunk = NULL; void *tptr = NULL; void *vcchunk = NULL;
1854         CatalogTableTABLE cTable(systemDatabase_);
1855         rv = cTable.getChunkAndTblPtr(elem->name, chunk, tptr, vcchunk);
1856         struct Object obj;
1857         strcpy(obj.name, elem->name);
1858         obj.type = Tbl;
1859         obj.bucketChunk = NULL;
1860         obj.firstPage = ((Chunk *)chunk)->getFirstPage();
1861         obj.curPage = ((Chunk *)chunk)->getCurrentPage();
1862         void *buf = &obj;
1863         write(fd, buf, sizeof(obj));
1864         FieldInfo *info = new FieldInfo();
1865         List fNameList = table->getFieldNameList();
1866         ListIterator fNameIter = fNameList.getIterator();
1867         count++;
1868         bool firstField=true;
1869         char fieldName[IDENTIFIER_LENGTH];
1870         while (fNameIter.hasElement()) {
1871             elem = (Identifier*) fNameIter.nextElement();
1872             Table::getFieldNameAlone(elem->name, fieldName);
1873             rv = table->getFieldInfo(elem->name, info);
1874             if (rv !=OK)  {
1875                 printf("unable to retrive info for table %s\n", elem->name);
1876             }
1877             if (firstField) {
1878                 fprintf(fp, "%s %s ", fieldName, AllDataType::getSQLString(info->type));
1879                 firstField = false;
1880             } else
1881                 fprintf(fp, ", %s %s ", fieldName, AllDataType::getSQLString(info->type));
1882             if (info->type == typeString) fprintf(fp, "(%d)",info->length );
1883             if (info->type == typeBinary) fprintf(fp, "(%d)",info->length);
1884             if (info->isNull) fprintf(fp, " NOT NULL ");
1885             if (info->isDefault) fprintf(fp, " DEFAULT '%s' ", info->defaultValueBuf);
1886             if (info->isAutoIncrement) fprintf(fp, " AUTO_INCREMENT ");
1887         }
1888         fprintf(fp, ");\n");
1889         table->printSQLIndexString(fp, fd);
1890         delete info;
1891         closeTable(table);
1892     }
1893     fclose(fp);
1894     close(fd);
1895     return OK;
1896 }
1897
1898 DbRetVal DatabaseManagerImpl::recover()
1899 {
1900     DbRetVal rv = OK;
1901     rv = sysDb()->recoverSystemDB();
1902     if (!Conf::config.useMmap())rv = db()->recoverUserDB();
1903     return rv;
1904 }
1905
1906 void DatabaseManagerImpl::sendSignal(int signal)
1907 {
1908     ThreadInfo* tInfo = sysDb()->getThreadInfo(0);
1909     for (int i=0; i < Conf::config.getMaxProcs(); i++)
1910     {
1911         if (tInfo->pid_ !=0) os::kill(tInfo->pid_, signal);
1912         tInfo++;
1913     }
1914 }