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