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