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