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