src/storage/DatabaseManagerImpl.cxx

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