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