src/server/DatabaseManagerImpl.c

   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
  29 DatabaseManagerImpl::~DatabaseManagerImpl()
  30 {
  31     //Note:Databases are closed by the session interface
  32     delete tMgr_;
  33     delete lMgr_;
  34 }
  35
  36 void DatabaseManagerImpl::createLockManager()
  37 {
  38     lMgr_ = new LockManager(systemDatabase_);
  39     return;
  40 }
  41
  42 void DatabaseManagerImpl::createTransactionManager()
  43 {
  44
  45     tMgr_ = new TransactionManager();
  46     tMgr_->setFirstTrans(systemDatabase_->getSystemDatabaseTrans(0));
  47     return;
  48 }
  49
  50 DbRetVal DatabaseManagerImpl::openSystemDatabase()
  51 {
  52     openDatabase(SYSTEMDB);
  53     systemDatabase_ = db_;
  54     db_ = NULL;
  55     if (NULL == systemDatabase_)
  56     {
  57         printError(ErrAlready, "Database is already opened");
  58         return ErrAlready;
  59     }
  60     printDebug(DM_Database, "Opened system database");
  61     logFinest(logger, "Opened system database");
  62     return OK;
  63 }
  64
  65 DbRetVal DatabaseManagerImpl::closeSystemDatabase()
  66 {
  67     Database *db  = db_;
  68     //make them to point to system database file descriptor
  69     //and database pointer
  70     db_ = systemDatabase_;
  71     closeDatabase();
  72     db_ = db;
  73     printDebug(DM_Database, "Closed system database");
  74     logFinest(logger, "Closed System database");
  75     return OK;
  76 }
  77
  78 DbRetVal DatabaseManagerImpl::createDatabase(const char *name, size_t size)
  79 {
  80     if (NULL != db_ )
  81     {
  82         printError(ErrAlready, "Database is already created");
  83         return ErrAlready;
  84     }
  85     caddr_t rtnAddr = (caddr_t) NULL;
  86     shared_memory_id shm_id = 0;
  87
  88     char *startaddr = (char*)config.getMapAddress();
  89     shared_memory_key key = 0;
  90     if (0 == strcmp(name, SYSTEMDB))
  91     {
  92
  93         key = config.getSysDbKey();
  94     }
  95     else
  96     {
  97         startaddr = startaddr + size;
  98         key = config.getUserDbKey();
  99     }
 100     shm_id = os::shm_create(key, size, 0666);
 101     if (-1 == shm_id)
 102     {
 103         printError(ErrOS, "Shared memory create failed");
 104         return ErrOS;
 105     }
 106
 107     void *shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 108     rtnAddr  = (caddr_t) shm_ptr;
 109     memset(shm_ptr, 0, size );
 110     if (rtnAddr < 0)
 111     {
 112         printError(ErrOS, "Shared memory attach returned -ve value %d", rtnAddr);
 113         return ErrOS;
 114     }
 115     db_ = new Database();
 116     printDebug(DM_Database, "Creating database:%s",name);
 117
 118     //TODO:for user database do not have transtable and processtable mutex
 119     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 120     db_->setDatabaseID(1);
 121     db_->setName(name);
 122     db_->setMaxSize(size);
 123     db_->setMaxChunks(MAX_CHUNKS);
 124     db_->initAllocDatabaseMutex();
 125     db_->initTransTableMutex();
 126     db_->initDatabaseMutex();
 127     db_->initProcessTableMutex();
 128
 129     //compute the first page after book keeping information
 130     size_t offset = os::alignLong(sizeof (DatabaseMetaData));
 131     //Only for system db chunk array, trans array and proc array will be there
 132     if (0 == strcmp(name, SYSTEMDB))
 133     {
 134         offset = offset + os::alignLong( MAX_CHUNKS  * sizeof (Chunk));
 135         offset = offset + os::alignLong( config.getMaxTrans()   * sizeof(Transaction));
 136         offset = offset + os::alignLong( config.getMaxProcs() * sizeof(int));
 137     }
 138     int multiple = os::floor(offset / PAGE_SIZE);
 139     char *curPage = (((char*)rtnAddr) + ((multiple + 1) * PAGE_SIZE));
 140
 141     db_->setCurrentPage(curPage);
 142     db_->setFirstPage(curPage);
 143
 144     if (0 == strcmp(name, SYSTEMDB)) return OK;
 145
 146     //Allocate new chunk to store hash index nodes
 147     Chunk *chunkInfo = createUserChunk(sizeof(HashIndexNode));
 148     if (NULL == chunkInfo)
 149     {
 150         printError(ErrSysInternal, "Failed to allocate hash index nodes chunk");
 151         return ErrSysInternal;
 152     }
 153     printDebug(DM_Database, "Creating Chunk for storing Hash index nodes %x",
 154                                                                   chunkInfo);
 155
 156     db_->setHashIndexChunk(chunkInfo);
 157     logFinest(logger, "Created database %s" , name);
 158
 159     return OK;
 160 }
 161
 162 DbRetVal DatabaseManagerImpl::deleteDatabase(const char *name)
 163 {
 164     shared_memory_id shm_id = 0;
 165     if (0 == strcmp(name, SYSTEMDB))
 166     {
 167         shm_id = os::shm_open(config.getSysDbKey(), 100, 0666);
 168         os::shmctl(shm_id, IPC_RMID);
 169     } else {
 170         shm_id = os::shm_open(config.getUserDbKey(), 100, 0666);
 171         os::shmctl(shm_id, IPC_RMID);
 172     }
 173     logFinest(logger, "Deleted database %s" , name);
 174     return OK;
 175 }
 176
 177
 178 DbRetVal DatabaseManagerImpl::openDatabase(const char *name)
 179 {
 180     size_t size = config.getMaxSysDbSize();
 181     char *startaddr = (char*)config.getMapAddress();
 182     if (0 == strcmp(name , SYSTEMDB))
 183     {
 184         if (NULL !=systemDatabase_)
 185         {
 186             printError(ErrAlready, "System Database already open");
 187             return ErrAlready;
 188         }
 189     }
 190     else
 191     {
 192
 193         if (NULL ==systemDatabase_)
 194         {
 195             printError(ErrNotOpen, "System Database not open");
 196             return ErrNotOpen;
 197         }
 198         size = config.getMaxDbSize();
 199         startaddr = startaddr + size;
 200     }
 201     if (NULL != db_)
 202     {
 203         printError(ErrAlready, "User Database already open");
 204         return ErrAlready;
 205     }
 206     //system db should be opened before user database files
 207
 208     caddr_t rtnAddr = (caddr_t) NULL;
 209     shared_memory_id shm_id = 0;
 210
 211     shared_memory_key key = 0;
 212     if (0 == strcmp(name, SYSTEMDB))
 213         key = config.getSysDbKey();
 214     else
 215         key = config.getUserDbKey();
 216     shm_id = os::shm_open(key, size, 0666);
 217     if (shm_id == -1)
 218     {
 219          printError(ErrOS, "Shared memory open failed");
 220          return ErrOS;
 221     }
 222
 223     void *shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 224
 225     rtnAddr  = (caddr_t) shm_ptr;
 226
 227     if (rtnAddr < 0)
 228     {
 229          printError(ErrOS, "Shared memory attach returned -ve value %d", rtnAddr);
 230          return ErrOS;
 231     }
 232     db_ = new Database();
 233     printDebug(DM_Database, "Opening database: %s", name);
 234     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 235     logFinest(logger, "Opened database %s" , name);
 236     return OK;
 237 }
 238
 239 void DatabaseManagerImpl::closeDatabase()
 240 {
 241     printDebug(DM_Database, "Closing database: %s",(char*)db_->getName());
 242     logFinest(logger, "Closed database");
 243     os::shm_detach((char*)db_->getMetaDataPtr());
 244     delete db_;
 245     db_ = NULL;
 246
 247 }
 248 //Assumes that system database mutex is taken before calling this.
 249 Chunk* DatabaseManagerImpl::createUserChunk(size_t size)
 250 {
 251     //Allocate new node in system database to store
 252     Chunk *chunk = getSystemTableChunk(UserChunkTableId);
 253     void *ptr = chunk->allocate(systemDatabase_);
 254     if (NULL == ptr)
 255     {
 256          printError(ErrNoMemory, "Allocation failed for User chunk catalog table");
 257          return NULL;
 258     }
 259     Chunk *chunkInfo = (Chunk*)ptr;
 260     chunkInfo->initMutex();
 261     if (size > PAGE_SIZE)
 262         chunkInfo->curPage_ = db_->getFreePage(size);
 263     else
 264         chunkInfo->curPage_ = db_->getFreePage();
 265     if ( NULL == chunkInfo->curPage_)
 266     {
 267         chunkInfo->destroyMutex();
 268         chunk->free(db_, ptr);
 269         printError(ErrNoMemory, "Database full: No space to allocate from database");
 270         return NULL;
 271     }
 272     PageInfo* firstPageInfo = ((PageInfo*)chunkInfo->curPage_);
 273     if (size > PAGE_SIZE)
 274     {
 275         int multiple = os::floor(size / PAGE_SIZE);
 276         int offset = ((multiple + 1) * PAGE_SIZE);
 277         firstPageInfo->setPageAsUsed(offset);
 278     }
 279     else
 280         firstPageInfo->setPageAsUsed(size);
 281     if (0 == size)
 282     {
 283         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
 284         varInfo->isUsed_ = 0;
 285         varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
 286
 287     }
 288     chunkInfo->firstPage_ = chunkInfo->curPage_;
 289
 290     if (0 == size)
 291         chunkInfo->setAllocType(VariableSizeAllocator);
 292     else
 293         chunkInfo->setAllocType(FixedSizeAllocator);
 294     chunkInfo->setSize(size);
 295
 296     //TODO::Generate chunkid::use tableid
 297     chunkInfo->setChunkID(-1);
 298     printDebug(DM_Database, "Creating new User chunk chunkID:%d size: %d firstPage:%x",
 299                                -1, size, firstPageInfo);
 300
 301     return chunkInfo;
 302 }
 303
 304
 305 //Assumes that system database mutex is taken before calling this.
 306 DbRetVal DatabaseManagerImpl::deleteUserChunk(Chunk *chunk)
 307 {
 308     //Go to the pages and set them to notUsed
 309     Page *page = chunk->firstPage_;
 310     PageInfo* pageInfo = ((PageInfo*)page);
 311     //Here...sure that atleast one page will be there even no tuples
 312     //are inserted.so not checking if pageInfo == NULL
 313     while( pageInfo->nextPage_ != NULL)
 314     {
 315         PageInfo *prev = pageInfo;
 316         pageInfo = (PageInfo*)(pageInfo->nextPage_);
 317         //sets pageInfo->isUsed_ = 0 and pageInfo->hasFreeSpace_ = 0
 318         //and initializes the page content to zero
 319         if(NULL == pageInfo->nextPageAfterMerge_)
 320             os::memset(prev, 0, PAGE_SIZE);
 321         else
 322         {
 323             int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 324             os::memset(prev, 0, size);
 325         }
 326         printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, prev);
 327     }
 328     //The above loop wont execute for the last page
 329     //and for the case where table has only one page
 330     if(NULL == pageInfo->nextPageAfterMerge_)
 331         os::memset(pageInfo, 0, PAGE_SIZE);
 332     else
 333     {
 334         int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 335         os::memset(pageInfo, 0, size);
 336     }
 337     printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, pageInfo);
 338     chunk->chunkID_ = -1;
 339     chunk->allocSize_ = 0;
 340     chunk->curPage_ = NULL;
 341     chunk->firstPage_ = NULL;
 342     chunk->destroyMutex();
 343     printDebug(DM_Database,"deleting user chunk:%x",chunk);
 344     return OK;
 345 }
 346
 347 //-1 -> Unable to create chunk. No memory
 348 //-2 -> Unable to update the catalog tables
 349 DbRetVal DatabaseManagerImpl::createTable(const char *name, TableDef &def)
 350 {
 351     DbRetVal rv = OK;
 352     size_t sizeofTuple = def.getTupleSize();
 353     rv = systemDatabase_->getDatabaseMutex();
 354     if (OK != rv ) {
 355         printError(rv, "Unable to get Database mutex");
 356         return rv;
 357     }
 358     //create a chunk to store the tuples
 359     Chunk *ptr = createUserChunk(sizeofTuple);
 360     if (NULL  == ptr)
 361     {
 362         systemDatabase_->releaseDatabaseMutex();
 363         printError(ErrNoResource, "Unable to create user chunk");
 364         return ErrNoResource;
 365     }
 366     printDebug(DM_Database,"Created UserChunk:%x", ptr);
 367     //add row to TABLE
 368     CatalogTableTABLE cTable(systemDatabase_);
 369     int tblID = ((Chunk*)ptr)->getChunkID();
 370     void *tptr = NULL;
 371     rv = cTable.insert(name, tblID, sizeofTuple,
 372                                    def.getFieldCount(), ptr, tptr);
 373     if (OK != rv)
 374     {
 375         deleteUserChunk(ptr);
 376         systemDatabase_->releaseDatabaseMutex();
 377         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 378         return ErrSysInternal;
 379     }
 380     printDebug(DM_Database,"Inserted into TABLE:%s",name);
 381     //add rows to FIELD
 382     FieldIterator iter = def.getFieldIterator();
 383     CatalogTableFIELD cField(systemDatabase_);
 384     rv = cField.insert(iter, tblID ,tptr);
 385     if (OK != rv)
 386     {
 387         deleteUserChunk(ptr);
 388         void *cptr, *ttptr;//Dummy as remove below needs both these OUT params
 389         cTable.remove(name, cptr, ttptr);
 390         systemDatabase_->releaseDatabaseMutex();
 391         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 392         return ErrSysInternal;
 393     }
 394     printDebug(DM_Database,"Inserted into FIELD:%s",name);
 395     systemDatabase_->releaseDatabaseMutex();
 396     printDebug(DM_Database,"Table Created:%s",name);
 397     logFinest(logger, "Table Created %s" , name);
 398     return OK;
 399 }
 400
 401 //TODO::If any operation fails in between, then we may have some
 402 //dangling tuples, say we have have rows in INDEX table
 403 //which will not have any corresponding entries in TABLE
 404 //CHANGE the sequence so that it deletes from the bottom as
 405 //opposed to start from top as is written now
 406 DbRetVal DatabaseManagerImpl::dropTable(const char *name)
 407 {
 408     void *chunk = NULL;
 409     void *tptr =NULL;
 410     systemDatabase_->getDatabaseMutex();
 411     //remove the entry in TABLE
 412     CatalogTableTABLE cTable(systemDatabase_);
 413     DbRetVal rv = cTable.remove(name, chunk, tptr);
 414     if (OK != rv) {
 415         systemDatabase_->releaseDatabaseMutex();
 416         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 417         return ErrSysInternal;
 418     }
 419     printDebug(DM_Database,"Deleted from TABLE:%s",name);
 420
 421     //remove the entries in the FIELD table
 422     CatalogTableFIELD cField(systemDatabase_);
 423     rv = cField.remove(tptr);
 424     if (OK != rv) {
 425         systemDatabase_->releaseDatabaseMutex();
 426         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 427         return ErrSysInternal;
 428     }
 429     printDebug(DM_Database,"Deleted from FIELD:%s",name);
 430
 431     rv = deleteUserChunk((Chunk*)chunk);
 432     if (OK != rv) {
 433         systemDatabase_->releaseDatabaseMutex();
 434         printError(rv, "Unable to delete the chunk");
 435         return rv;
 436     }
 437     printDebug(DM_Database,"Deleted UserChunk:%x", chunk);
 438
 439     //TODO::check whether indexes are available and drop that also.
 440     CatalogTableINDEX cIndex(systemDatabase_);
 441     int noIndexes = cIndex.getNumIndexes(tptr);
 442     char *idxName;
 443     for (int i =1 ; i<= noIndexes; i++) {
 444         idxName = cIndex.getIndexName(tptr, i);
 445         dropIndex(idxName);
 446     }
 447     systemDatabase_->releaseDatabaseMutex();
 448     printDebug(DM_Database, "Deleted Table %s" , name);
 449     logFinest(logger, "Deleted Table %s" , name);
 450     return OK;
 451 }
 452
 453 //Return values: NULL for table not found
 454 Table* DatabaseManagerImpl::openTable(const char *name)
 455 {
 456     DbRetVal ret = OK;
 457     //TODO::store table handles in list so that if it is
 458     //not closed by the application. destructor shall close it.
 459     TableImpl *table = new TableImpl();
 460     table->setDB(db_);
 461     table->setSystemDB(systemDatabase_);
 462     table->setLockManager(lMgr_);
 463     table->setTrans(&(tMgr_->trans));
 464
 465     //to store the chunk pointer of table
 466     void *chunk = NULL;
 467
 468     //to store the tuple pointer of the table
 469     void *tptr =NULL;
 470
 471     //TODO::need to take shared lock on the table so that
 472     //all ddl operation will be denied on that table
 473     //which includes index creation, alter table
 474
 475     systemDatabase_->getDatabaseMutex();
 476     CatalogTableTABLE cTable(systemDatabase_);
 477     ret = cTable.getChunkAndTblPtr(name, chunk, tptr);
 478     if ( OK != ret)
 479     {
 480         printError(ErrNotExists, "Table not exists %s", name);
 481         return NULL;
 482     }
 483     TABLE *tTuple = (TABLE*)tptr;
 484     table->setTableInfo(tTuple->tblName_, tTuple->tblID_, tTuple->length_,
 485                         tTuple->numFlds_, tTuple->numIndexes_, tTuple->chunkPtr_);
 486
 487     //get field information from FIELD table
 488     CatalogTableFIELD cField(systemDatabase_);
 489     cField.getFieldInfo(tptr, table->fldList_);
 490
 491     //get the number of indexes on this table
 492     //and populate the indexPtr array
 493     CatalogTableINDEX cIndex(systemDatabase_);
 494     table->numIndexes_ = cIndex.getNumIndexes(tptr);
 495     if (table->numIndexes_)
 496         table->indexPtr_ = new char*[table->numIndexes_];
 497     else
 498         table->indexPtr_ = NULL;
 499     cIndex.getIndexPtrs(tptr, table->indexPtr_);
 500
 501     //HACK:Below works only for one hash index on table
 502     if (table->numIndexes_ == 1)
 503     {
 504         SingleFieldHashIndexInfo *hIdxInfo = new SingleFieldHashIndexInfo();
 505         CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 506         cIndexField.getFieldNameAndType(table->indexPtr_[0], hIdxInfo->fldName,
 507                                                       hIdxInfo->type);
 508         ChunkIterator citer = CatalogTableINDEX::getIterator(table->indexPtr_[0]);
 509         hIdxInfo->noOfBuckets = CatalogTableINDEX::getNoOfBuckets(table->indexPtr_[0]);
 510         hIdxInfo->buckets = (Bucket*)citer.nextElement();
 511
 512         table->idxInfo = (IndexInfo*) hIdxInfo;
 513     }
 514     systemDatabase_->releaseDatabaseMutex();
 515     printDebug(DM_Database,"Opening table handle name:%s chunk:%x numIndex:%d",
 516                                          name, chunk, table->numIndexes_);
 517     logFinest(logger, "Opening Table %s" , name);
 518
 519     return table;
 520 }
 521
 522 //Return values: -1 for table not found
 523 void DatabaseManagerImpl::closeTable(Table *table)
 524 {
 525     printDebug(DM_Database,"Closing table handle: %x", table);
 526     if (NULL == table) return;
 527     delete table;
 528     logFinest(logger, "Closing Table");
 529 }
 530
 531 DbRetVal DatabaseManagerImpl::createIndex(const char *indName, IndexInitInfo *info)
 532 {
 533     DbRetVal rv = OK;
 534     if (info->indType == hashIndex)
 535     {
 536         //Assumes info is of type HashIndexInitInfo
 537         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
 538         rv = createHashIndex(indName, info->tableName, info->list, hInfo->bucketSize);
 539     }
 540     else
 541     {
 542         //TODO::tree index
 543     }
 544     return rv;
 545 }
 546
 547
 548 //-1 -> Table does not exists
 549 //-2 -> Field does not exists
 550 //-3 -> bucketSize is not valid
 551 DbRetVal DatabaseManagerImpl::createHashIndex(const char *indName, const char *tblName,
 552                       FieldNameList &fldList, int bucketSize)
 553 {
 554     //validate the bucket size
 555     if (bucketSize < 100 || bucketSize > 200000)
 556     {
 557         printError(ErrBadRange, "Index Bucket size %d not in range 100-200000",
 558                                  bucketSize);
 559         return ErrBadRange;
 560     }
 561     void *tptr =NULL;
 562     void *chunk = NULL;
 563     systemDatabase_->getDatabaseMutex();
 564
 565     //check whether table exists
 566     CatalogTableTABLE cTable(systemDatabase_);
 567     cTable.getChunkAndTblPtr(tblName, chunk, tptr);
 568     if (NULL == tptr)
 569     {
 570         systemDatabase_->releaseDatabaseMutex();
 571         printError(ErrNotExists, "Table does not exist %s", tblName);
 572         return ErrNotExists;
 573     }
 574     int totFlds = fldList.size();
 575
 576     //check whether field exists
 577     char **fptr = new char* [totFlds];
 578     CatalogTableFIELD cField(systemDatabase_);
 579     DbRetVal rv = cField.getFieldPtrs(fldList, tptr, fptr);
 580     if (OK != rv)
 581     {
 582         delete[] fptr;
 583         systemDatabase_->releaseDatabaseMutex();
 584         printError(ErrNotExists, "Field does not exist");
 585         return ErrNotExists;
 586     }
 587
 588     //create chunk to store the meta data of the index created
 589     //for latches and bucket pointers
 590     Chunk* chunkInfo = createUserChunk(bucketSize * sizeof(Bucket));
 591     if (NULL == chunkInfo)
 592     {
 593         delete[] fptr;
 594         systemDatabase_->releaseDatabaseMutex();
 595         printError(ErrSysInternal, "Unable to create chunk");
 596         return ErrSysInternal;
 597     }
 598     //create memory for holding the bucket pointers
 599     void *buckets = chunkInfo->allocate(db_);
 600     if (NULL == buckets)
 601     {
 602         delete[] fptr;
 603         deleteUserChunk(chunkInfo);
 604         systemDatabase_->releaseDatabaseMutex();
 605         printError(ErrNoMemory, "Unable to allocate memory for bucket");
 606         return ErrNoMemory;
 607     }
 608     Bucket *buck = (Bucket*) buckets;
 609     initHashBuckets(buck, bucketSize);
 610
 611     //create chunk to store the hash index nodes
 612     Chunk* hChunk = createUserChunk(sizeof(HashIndexNode));
 613     if (NULL == hChunk)
 614     {
 615         delete[] fptr;
 616         deleteUserChunk(chunkInfo);
 617         systemDatabase_->releaseDatabaseMutex();
 618         printError(ErrSysInternal, "Unable to create chunk for storing hash index nodes");
 619         return ErrSysInternal;
 620     }
 621
 622
 623     //add row to INDEX
 624     void *tupleptr = NULL;
 625     CatalogTableINDEX cIndex(systemDatabase_);
 626     DbRetVal ret = cIndex.insert(indName, tptr, fldList.size(),
 627                         chunkInfo, bucketSize, hChunk, tupleptr);
 628     if (OK != rv)
 629     {
 630         delete[] fptr;
 631         deleteUserChunk(hChunk);
 632         deleteUserChunk(chunkInfo);
 633         systemDatabase_->releaseDatabaseMutex();
 634         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
 635         return ErrSysInternal;
 636     }
 637     //add rows to INDEXFIELD
 638     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 639     ret = cIndexField.insert(fldList, tupleptr, tptr, fptr);
 640
 641     if (OK != rv)
 642     {
 643         delete[] fptr;
 644         deleteUserChunk(hChunk);
 645         deleteUserChunk(chunkInfo);
 646         systemDatabase_->releaseDatabaseMutex();
 647         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
 648         return ErrSysInternal;
 649     }
 650     delete[] fptr;
 651     //TODO::If tuples present in this table, then
 652     //create hash index nodes and store it
 653     //Take table lock
 654     systemDatabase_->releaseDatabaseMutex();
 655     printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x",
 656                                    indName, tblName, buckets);
 657     logFinest(logger, "Creating HashIndex %s on %s with bucket size %d",
 658                                    indName, tblName, buckets);
 659     return OK;
 660 }
 661
 662 void DatabaseManagerImpl::initHashBuckets(Bucket *buck, int bucketSize)
 663 {
 664     os::memset((void*)buck, 0, bucketSize * sizeof(Bucket));
 665
 666     for (int i=0; i < bucketSize ; i++)
 667     {
 668         buck[i].mutex_.init();
 669     }
 670     return;
 671 }
 672
 673 DbRetVal DatabaseManagerImpl::dropIndex(const char *name)
 674 {
 675     DbRetVal rv = OK;
 676     void *chunk = NULL, *hchunk = NULL;
 677     void *tptr =NULL;
 678     int ret = 0;
 679     systemDatabase_->getDatabaseMutex();
 680     //remove the entry in INDEX
 681     CatalogTableINDEX cIndex(systemDatabase_);
 682     rv = cIndex.remove(name, chunk, hchunk, tptr);
 683     if (OK != rv)
 684     {
 685         systemDatabase_->releaseDatabaseMutex();
 686         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
 687         return ErrSysInternal;
 688     }
 689     printDebug(DM_Database, "Removing from INDEX %s",name);
 690     //remove the entries in the INDEXFIELD table
 691     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 692     rv = cIndexField.remove(tptr);
 693     if (OK != rv)
 694     {
 695         systemDatabase_->releaseDatabaseMutex();
 696         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
 697         return ErrSysInternal;
 698     }
 699     printDebug(DM_Database, "Removing from INDEXFIELD %s",name);
 700
 701     //delete the index chunk
 702     rv = deleteUserChunk((Chunk*)chunk);
 703     if (OK != rv)
 704     {
 705         systemDatabase_->releaseDatabaseMutex();
 706         printError(ErrSysInternal, "Unable to delete the index chunk");
 707         return ErrSysInternal;
 708     }
 709     //delete the index hash node chunk
 710     rv = deleteUserChunk((Chunk*)hchunk);
 711     if (OK != rv)
 712     {
 713         systemDatabase_->releaseDatabaseMutex();
 714         printError(ErrSysInternal, "Unable to delete the index hash node chunk");
 715         return ErrSysInternal;
 716     }
 717     systemDatabase_->releaseDatabaseMutex();
 718     //TODO::If tuples present in this table, then
 719     //free all hash index nodes for this table.
 720     //free all nodes in list of all buckets
 721     //Take table lock
 722     printDebug(DM_Database, "Dropped hash index %s",name);
 723     logFinest(logger, "Deleted Index %s", name);
 724     return OK;
 725 }
 726
 727
 728
 729 ChunkIterator DatabaseManagerImpl::getSystemTableIterator(CatalogTableID id)
 730 {
 731     Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(id);
 732     return fChunk->getIterator();
 733 }
 734
 735 Chunk* DatabaseManagerImpl::getSystemTableChunk(CatalogTableID id)
 736 {
 737     return systemDatabase_->getSystemDatabaseChunk(id);
 738 }