src/storage/DbMgrIndexImpl.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<Process.h>
  29
  30 DbRetVal DatabaseManagerImpl::renameIndex(const char *oldName,const char *newName)
  31 {
  32     void *chunk = NULL;
  33     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
  34     if (OK != rv) {
  35         printError(ErrSysInternal, "Unable to get database mutex for rename table");
  36         return ErrSysInternal;
  37     }
  38     CatalogTableTABLE cTable(systemDatabase_);
  39     rv = cTable.renameIndex(oldName,newName);
  40     if (OK != rv) {
  41         printError(ErrSysInternal, "Unable to rename table");
  42         systemDatabase_->releaseCheckpointMutex();
  43         return ErrSysInternal;
  44     }
  45     systemDatabase_->releaseCheckpointMutex();
  46     return OK;
  47 }
  48
  49 DbRetVal DatabaseManagerImpl::createIndex(const char *indName, IndexInitInfo *info)
  50 {
  51     DbRetVal rv = OK;
  52     if (!info->isUnique && info->isPrimary)
  53     {
  54         printError(ErrBadCall, "Primary key cannot be non unique\n");
  55         return ErrBadCall;
  56     }
  57     if (!Util::isIdentifier((char*)indName)) {
  58         printError(ErrBadArg, "Invalid character for index name");
  59         return ErrBadArg;
  60     }
  61
  62     if (info->indType == hashIndex)
  63     {
  64         //Assumes info is of type HashIndexInitInfo
  65         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
  66         rv = createHashIndex(indName, info->tableName, info->list, hInfo->bucketSize,
  67                              info->isUnique, info->isPrimary);
  68     }
  69     else if (info->indType == treeIndex)
  70     {
  71         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
  72         rv = createTreeIndex(indName, info->tableName, info->list,
  73                           hInfo->bucketSize, info->isUnique, info->isPrimary);
  74     }
  75     else if (info->indType == trieIndex)
  76     {
  77         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
  78         rv = createTrieIndex(indName, info->tableName, info->list,
  79                           info->isUnique, info->isPrimary);
  80
  81     }else {
  82         printError(ErrBadCall, "Index type not supported\n");
  83         return ErrBadCall;
  84     }
  85     return rv;
  86 }
  87
  88
  89 //-1 -> Table does not exists
  90 //-2 -> Field does not exists
  91 //-3 -> bucketSize is not valid
  92 DbRetVal DatabaseManagerImpl::createHashIndex(const char *indName, const char *tblName,
  93                       FieldNameList &fldList, int bucketSize, bool isUnique, bool isPrimary)
  94 {
  95     //validate the bucket size
  96     if (bucketSize < 100 || bucketSize > 200000)
  97     {
  98         printError(ErrBadRange, "Index Bucket size %d not in range 100-200000",
  99                                  bucketSize);
 100         return ErrBadRange;
 101     }
 102     int totFlds = fldList.size();
 103     if (totFlds == 0)
 104     {
 105         printError(ErrBadCall, "No Field name specified");
 106         return ErrBadCall;
 107     }
 108     void *tptr =NULL;
 109     void *chunk = NULL;
 110     void *vcchunk = NULL;
 111     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 112     if (OK != rv)
 113     {
 114         printError(ErrSysInternal, "Unable to get database mutex");
 115         return ErrSysInternal;
 116     }
 117
 118     //check whether table exists
 119     CatalogTableTABLE cTable(systemDatabase_);
 120     cTable.getChunkAndTblPtr(tblName, chunk, tptr, vcchunk);
 121     if (NULL == tptr)
 122     {
 123         systemDatabase_->releaseCheckpointMutex();
 124         printError(ErrNotExists, "Table does not exist %s", tblName);
 125         return ErrNotExists;
 126     }
 127
 128     //check whether field exists
 129     char **fptr = new char* [totFlds];
 130     CatalogTableFIELD cField(systemDatabase_);
 131     rv = cField.getFieldPtrs(fldList, tptr, fptr);
 132     if (OK != rv)
 133     {
 134         delete[] fptr;
 135         systemDatabase_->releaseCheckpointMutex();
 136         //TODO::check test cases of dbapi/Index, they give wrong results
 137         //if (rv == ErrBadCall) {
 138         ////    if (isPrimary) printError(ErrBadCall, "Field can have NULL values");
 139         //} else {
 140         //printError(ErrNotExists, "Field does not exist");
 141         //}
 142         //return ErrBadCall;
 143         if (rv != ErrBadCall) {
 144             printError(ErrNotExists, "Field does not exist");
 145             return ErrNotExists;
 146         }
 147     }
 148     for (int i=0; i <totFlds; i++)
 149     {
 150         CFIELD* fInfo = (CFIELD*)fptr[i];
 151         if (fInfo->type_ == typeFloat || fInfo->type_ == typeDouble || fInfo->type_ == typeTimeStamp)
 152         {
 153             printError(ErrBadArg, "HashIndex cannot be created for float or double or timestamp type");
 154             delete[] fptr;
 155             systemDatabase_->releaseCheckpointMutex();
 156             return ErrBadArg;
 157         }
 158         if (!fInfo->isNull_ && isPrimary )
 159         {
 160             printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
 161             delete[] fptr;
 162             systemDatabase_->releaseCheckpointMutex();
 163             return ErrBadArg;
 164         }
 165          if(isPrimary){fInfo->isPrimary_=true;fInfo->isUnique_=true;}
 166          if(isUnique){fInfo->isUnique_=true;}
 167     }
 168     //create chunk to store the meta data of the index created
 169     //for latches and bucket pointers
 170     printDebug(DM_HashIndex, "Creating chunk for storing hash buckets of size %d\n",
 171                              bucketSize * sizeof(Bucket));
 172     Chunk* chunkInfo = createUserChunk(bucketSize * sizeof(Bucket));
 173     if (NULL == chunkInfo)
 174     {
 175         delete[] fptr;
 176         systemDatabase_->releaseCheckpointMutex();
 177         printError(ErrSysInternal, "Unable to create chunk");
 178         return ErrSysInternal;
 179     }
 180     chunkInfo->setChunkName(indName);
 181     //create memory for holding the bucket pointers
 182     void *buckets = chunkInfo->allocate(db_, &rv);
 183     if (NULL == buckets)
 184     {
 185         delete[] fptr;
 186         deleteUserChunk(chunkInfo);
 187         systemDatabase_->releaseCheckpointMutex();
 188         printError(rv, "Unable to allocate memory for bucket");
 189         return rv;
 190     }
 191     Bucket *buck = (Bucket*) buckets;
 192     initHashBuckets(buck, bucketSize);
 193
 194     //create chunk to store the hash index nodes
 195     Chunk* hChunk = createUserChunk(sizeof(IndexNode));
 196     if (NULL == hChunk)
 197     {
 198         delete[] fptr;
 199         deleteUserChunk(chunkInfo);
 200         systemDatabase_->releaseCheckpointMutex();
 201         printError(ErrSysInternal, "Unable to create chunk for storing hash index nodes");
 202         return ErrSysInternal;
 203     }
 204
 205      hChunk->setChunkName(indName);
 206     //add row to INDEX
 207     void *tupleptr = NULL;
 208     CatalogTableINDEX cIndex(systemDatabase_);
 209     rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
 210                         chunkInfo, bucketSize, hChunk, tupleptr);
 211     if (OK != rv)
 212     {
 213         delete[] fptr;
 214         deleteUserChunk(hChunk);
 215         deleteUserChunk(chunkInfo);
 216         systemDatabase_->releaseCheckpointMutex();
 217         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
 218         return ErrSysInternal;
 219     }
 220     //add rows to INDEXFIELD
 221     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 222     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
 223
 224     if (OK != rv)
 225     {
 226         delete[] fptr;
 227         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
 228         deleteUserChunk(hChunk);
 229         deleteUserChunk(chunkInfo);
 230         systemDatabase_->releaseCheckpointMutex();
 231         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
 232         return ErrSysInternal;
 233     }
 234     delete[] fptr;
 235     systemDatabase_->releaseCheckpointMutex();
 236
 237     //TODO:: Take table lock
 238
 239     // Following code is written by Kishor Amballi
 240     TableImpl *tbl = (TableImpl *) openTable(tblName);
 241     if (NULL == tbl) {
 242         printError(ErrSysInternal, "Unable to open table %s", tblName);
 243         return ErrSysInternal;
 244     }
 245     if (! tbl->numTuples()) {
 246         printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x", indName, tblName, buckets);
 247         logFinest(Conf::logger, "Creating HashIndex %s on %s with bucket size %d", indName, tblName, buckets);
 248         closeTable(tbl);
 249         return OK;
 250     }
 251     HashIndexInfo *indxInfo = NULL;
 252     int i = 0;
 253     for (i = 0; i < tbl->numIndexes_; i++) {
 254         if(((HashIndexInfo *)tbl->idxInfo[i])->indexPtr == tupleptr) {
 255             indxInfo = (HashIndexInfo *) tbl->idxInfo[i];
 256             break;
 257         }
 258     }
 259     void *recPtr = NULL;
 260     ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
 261     tbl->setLoading(true);
 262     while ((recPtr = chIter.nextElement()) != NULL) {
 263         rv = tbl->insertIndexNode(*tbl->trans, tupleptr, indxInfo, recPtr);
 264         if (rv == ErrUnique) {
 265             closeTable(tbl);
 266             dropIndex(indName);
 267             return rv;
 268         }
 269     }
 270     closeTable(tbl);
 271     printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x", indName, tblName, buckets);
 272     logFinest(Conf::logger, "Creating HashIndex %s on %s with bucket size %d", indName, tblName, buckets);
 273     return OK;
 274 }
 275
 276
 277 DbRetVal DatabaseManagerImpl::createTreeIndex(const char *indName, const char *tblName,
 278                       FieldNameList &fldList, int nodeSize, bool isUnique, bool isPrimary)
 279 {
 280     if (nodeSize < 20 || nodeSize > 20000)
 281     {
 282         printError(ErrBadRange,"Tree Index Node size %d not in range 20-20000",
 283                                  nodeSize);
 284         return ErrBadRange;
 285     }
 286     int totFlds = fldList.size();
 287     if (totFlds == 0) {
 288         printError(ErrBadCall, "No Field name specified");
 289         return ErrBadCall;
 290     }else if (totFlds != 1) {
 291         printError(ErrBadCall, "Composite index not supported for Tree");
 292         return ErrBadCall;
 293     }
 294     void *tptr =NULL;
 295     void *chunk = NULL;
 296     void *vcchunk = NULL;
 297     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 298     if (OK != rv)
 299     {
 300         printError(ErrSysInternal, "Unable to get database mutex");
 301         return ErrSysInternal;
 302     }
 303     //check whether table exists
 304
 305     CatalogTableTABLE cTable(systemDatabase_);
 306     cTable.getChunkAndTblPtr(tblName, chunk, tptr, vcchunk);
 307     if (NULL == tptr)
 308     {
 309         systemDatabase_->releaseCheckpointMutex();
 310         printError(ErrNotExists, "Table does not exist %s", tblName);
 311         return ErrNotExists;
 312     }
 313     char **fptr = new char* [totFlds];
 314     CatalogTableFIELD cField(systemDatabase_);
 315     rv = cField.getFieldPtrs(fldList, tptr, fptr);
 316     if (OK != rv)
 317     {
 318         delete[] fptr;
 319         systemDatabase_->releaseCheckpointMutex();
 320         if (rv != ErrBadCall) {
 321             printError(ErrNotExists, "Field does not exist");
 322             return ErrNotExists;
 323         }
 324     }
 325     for (int i=0; i <totFlds; i++)
 326     {
 327         CFIELD* fInfo = (CFIELD*)fptr[i];
 328         if (!fInfo->isNull_ && isPrimary )
 329         {
 330             printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
 331             delete[] fptr;
 332             systemDatabase_->releaseCheckpointMutex();
 333             return ErrBadArg;
 334         }
 335         if (fInfo->type_ == typeVarchar)
 336         {
 337             printError(ErrBadArg, "Tree Index not supported for varchar type. Use char data type instead.");
 338             delete[] fptr;
 339            systemDatabase_->releaseCheckpointMutex();
 340             return ErrBadArg;
 341         }
 342
 343     }
 344     int chunkSize = sizeof(TreeNode)+(nodeSize * sizeof(void*));
 345     printDebug(DM_HashIndex, "Creating chunk for storing tree nodes of size %d\n", chunkSize);
 346
 347     Chunk* chunkInfo = createUserChunk(chunkSize);
 348     if (NULL == chunkInfo)
 349     {
 350         delete[] fptr;
 351         systemDatabase_->releaseCheckpointMutex();
 352         printError(ErrSysInternal, "Unable to create chunk");
 353         return ErrSysInternal;
 354     }
 355
 356
 357     void *tupleptr = NULL;
 358
 359     CatalogTableINDEX cIndex(systemDatabase_);
 360     rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
 361                         chunkInfo, nodeSize, NULL, tupleptr);
 362     if (OK != rv)
 363     {
 364         delete[] fptr;
 365         deleteUserChunk(chunkInfo);
 366         systemDatabase_->releaseCheckpointMutex();
 367         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
 368         return ErrSysInternal;
 369     }
 370     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 371     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
 372
 373     if (OK != rv)
 374     {
 375         delete[] fptr;
 376         void *hChunk = NULL;
 377         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
 378         deleteUserChunk(chunkInfo);
 379         systemDatabase_->releaseCheckpointMutex();
 380         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
 381         return ErrSysInternal;
 382     }
 383     delete[] fptr;
 384     rv = createIndexNodeForRecords(tblName, tupleptr, chunk);
 385     if (rv != OK)
 386     {
 387         //dropIndex again takes checkpoint mutex so releasing it here
 388         systemDatabase_->releaseCheckpointMutex();
 389         dropIndex(indName);
 390         return rv;
 391     }
 392     systemDatabase_->releaseCheckpointMutex();
 393     logFinest(Conf::logger, "Creating TreeIndex %s on %s rv:%d",
 394                                    indName, tblName, rv);
 395     return OK;
 396 }
 397 DbRetVal DatabaseManagerImpl::createIndexNodeForRecords(const char* tblName,
 398                               void *tupleptr, void *chunk)
 399 {
 400     //TODO::if tuples already present in this table, then create tree index '
 401     //nodes
 402     TableImpl *tbl = (TableImpl *) openTable(tblName);
 403     if (NULL == tbl) {
 404         printError(ErrSysInternal, "Unable to open table %s", tblName);
 405         return ErrSysInternal;
 406     }
 407     if (! tbl->numTuples()) {
 408         closeTable(tbl);
 409         return OK;
 410     }
 411     HashIndexInfo *indxInfo = NULL;
 412     int i = 0;
 413     for (i = 0; i < tbl->numIndexes_; i++) {
 414         if(((HashIndexInfo *)tbl->idxInfo[i])->indexPtr == tupleptr) {
 415             indxInfo = (HashIndexInfo *) tbl->idxInfo[i];
 416             break;
 417         }
 418     }
 419     void *recPtr = NULL;
 420     ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
 421     tbl->setLoading(true);
 422     DbRetVal rv = OK;
 423     while ((recPtr = chIter.nextElement()) != NULL) {
 424         rv = tbl->insertIndexNode(*tbl->trans, tupleptr, indxInfo, recPtr);
 425         if (rv == ErrUnique) {
 426             closeTable(tbl);
 427             return rv;
 428         }
 429     }
 430     closeTable(tbl);
 431     return OK;
 432 }
 433
 434 DbRetVal DatabaseManagerImpl::createTrieIndex(const char *indName, const char *tblName,
 435                       FieldNameList &fldList, bool isUnique, bool isPrimary)
 436 {
 437     int totFlds = fldList.size();
 438     void *tptr =NULL;
 439     char **fptr = new char* [totFlds];
 440     DbRetVal rv = validateIndex(tblName, fldList, &tptr, &fptr, isPrimary);
 441     if (OK != rv)
 442     {
 443         delete[] fptr;
 444         return rv;
 445     }
 446     rv = systemDatabase_->getXCheckpointMutex();
 447     if (OK != rv)
 448     {
 449         printError(ErrSysInternal, "Unable to get database mutex");
 450         return ErrSysInternal;
 451     }
 452
 453     //below statements are actually setting values in the catalog table
 454     //thats why mutex is taken before this stmt. Do not change the order
 455     CFIELD* fInfo = (CFIELD*)fptr[0];
 456     if(isPrimary){fInfo->isPrimary_=true;fInfo->isUnique_=true;}
 457     if(isUnique){fInfo->isUnique_=true;}
 458
 459     printDebug(DM_TrieIndex, "Creating chunk for storing trie nodes\n" );
 460     Chunk* chunkInfo = createUserChunk(sizeof(TrieNode));
 461
 462     //chunk to store the linked list of trie values
 463     Chunk* hChunk = createUserChunk(sizeof(IndexNode));
 464     if (NULL == chunkInfo || NULL == hChunk)
 465     {
 466         delete[] fptr;
 467         if (chunkInfo) deleteUserChunk(chunkInfo);
 468         systemDatabase_->releaseCheckpointMutex();
 469         printError(ErrSysInternal, "Unable to create trie node chunk");
 470         return ErrSysInternal;
 471     }
 472     chunkInfo->setChunkName(indName);
 473     hChunk->setChunkName(indName);
 474     void *tupleptr = NULL;
 475     rv = updateIndexCatalogTables(indName,tptr, fptr, fldList, isUnique,
 476                                   chunkInfo, hChunk , tupleptr);
 477     delete[] fptr;
 478     if (OK != rv) {
 479         printError(ErrSysInternal, "Catalog table updation failed");
 480         deleteUserChunk(chunkInfo);
 481         deleteUserChunk(hChunk);
 482         systemDatabase_->releaseCheckpointMutex();
 483         return rv;
 484     }
 485
 486     void *chunk = NULL;
 487     void *vcchunk = NULL;
 488     CatalogTableTABLE cTable(systemDatabase_);
 489     cTable.getChunkAndTblPtr(tblName, chunk, tptr, vcchunk);
 490     //create index nodes if records already exist in the table
 491     rv = createIndexNodeForRecords(tblName, tupleptr, chunk);
 492     if (rv != OK)
 493     {
 494         dropIndex(indName);
 495         systemDatabase_->releaseCheckpointMutex();
 496         return rv;
 497     }
 498     systemDatabase_->releaseCheckpointMutex();
 499     logFinest(Conf::logger, "Creating TrieIndex %s on %s rv:%d",
 500                                    indName, tblName, rv);
 501     return OK;
 502 }
 503 DbRetVal DatabaseManagerImpl::validateIndex(const char *tblName,
 504                              FieldNameList &fldList, void **tptr, char ***fptr,
 505                              bool isPrimary)
 506 {
 507     int totFlds = fldList.size();
 508     if (totFlds != 1)
 509     {
 510         printError(ErrBadCall, "No Field name specified or composite fields specified");
 511         return ErrBadCall;
 512     }
 513     void *chunk = NULL;
 514     void *vcchunk = NULL;
 515     //check whether table exists
 516     CatalogTableTABLE cTable(systemDatabase_);
 517     cTable.getChunkAndTblPtr(tblName, chunk, *tptr, vcchunk);
 518     if (NULL == tptr)
 519     {
 520         printError(ErrNotExists, "Table does not exist %s", tblName);
 521         return ErrNotExists;
 522     }
 523
 524     //check whether field exists
 525     CatalogTableFIELD cField(systemDatabase_);
 526     DbRetVal rv = cField.getFieldPtrs(fldList, *tptr, *fptr);
 527     if (OK != rv)
 528     {
 529         if (rv != ErrBadCall) {
 530             printError(ErrNotExists, "Field does not exist");
 531             return ErrNotExists;
 532         }
 533     }
 534     CFIELD* fInfo = (CFIELD*)*fptr[0];
 535     if (!(fInfo->type_ == typeInt || fInfo->type_ == typeLongLong || fInfo->type_ == typeString || fInfo->type_ == typeVarchar))
 536     {
 537         printError(ErrBadArg, "Trie Index cannot be created for float or double or timestamp type");
 538         return ErrBadArg;
 539     }
 540     if (!fInfo->isNull_ && isPrimary )
 541     {
 542         printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
 543         return ErrBadArg;
 544     }
 545
 546     return OK;
 547 }
 548
 549 DbRetVal DatabaseManagerImpl::updateIndexCatalogTables(const char *indName,
 550                         void *tptr, char **fptr, FieldNameList &fldList,
 551                         bool isUnique, Chunk* chunkInfo, Chunk* hChunk,
 552                         void *&tupleptr )
 553 {
 554     //void *tupleptr = NULL;
 555     CatalogTableINDEX cIndex(systemDatabase_);
 556     DbRetVal rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
 557                         chunkInfo, 0, hChunk, tupleptr);
 558     if (OK != rv)
 559     {
 560         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
 561         return ErrSysInternal;
 562     }
 563     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 564     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
 565     if (OK != rv)
 566     {
 567         //rollback the previous operation
 568         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
 569         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
 570         return ErrSysInternal;
 571     }
 572     return rv;
 573 }
 574
 575 DbRetVal DatabaseManagerImpl::removeIndexCatalogTables(const char *name, void *&chunk, void *&hchunk, void *&tptr)
 576 {
 577     //remove the entry in INDEX
 578     CatalogTableINDEX cIndex(systemDatabase_);
 579     DbRetVal rv = cIndex.remove(name, chunk, hchunk, tptr);
 580     if (OK != rv)
 581     {
 582         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
 583         return ErrSysInternal;
 584     }
 585     printDebug(DM_Database, "Removing from INDEX %s",name);
 586     //remove the entries in the INDEXFIELD table
 587     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 588     rv = cIndexField.remove(tptr);
 589     if (OK != rv)
 590     {
 591         printError(ErrSysInternal, "Catalog table updation failed for INDEX table");
 592         return ErrSysInternal;
 593     }
 594     printDebug(DM_Database, "Removing from INDEXFIELD %s",name);
 595     return OK;
 596 }
 597
 598 DbRetVal DatabaseManagerImpl::removeIndexChunks(void* chunk, void* hchunk, IndexType iType)
 599 {
 600     DbRetVal rv = deleteUserChunk((Chunk*)chunk);
 601     if (OK != rv)
 602     {
 603         printError(ErrSysInternal, "Unable to delete the index chunk");
 604         return ErrSysInternal;
 605     }
 606     //delete the index hash node chunk
 607     if (iType == hashIndex || iType == trieIndex) {
 608         rv = deleteUserChunk((Chunk*)hchunk);
 609         if (OK != rv)
 610         {
 611             printError(ErrSysInternal, "Unable to delete the index hash node chunk");
 612             return ErrSysInternal;
 613         }
 614     }
 615     return OK;
 616 }
 617
 618 void DatabaseManagerImpl::initHashBuckets(Bucket *buck, int bucketSize)
 619 {
 620     os::memset((void*)buck, 0, bucketSize * sizeof(Bucket));
 621     char mutName[IDENTIFIER_LENGTH];
 622     for (int i=0; i < bucketSize ; i++)
 623     {
 624         sprintf(mutName, "BKT:%d",i);
 625         buck[i].mutex_.init(mutName);
 626     }
 627     return;
 628 }
 629
 630 DbRetVal DatabaseManagerImpl::dropIndex(const char *name)
 631 {
 632     return dropIndexInt(name, true);
 633 }
 634
 635 DbRetVal DatabaseManagerImpl::dropIndexInt(const char *name, bool takeLock)
 636 {
 637     DbRetVal rv = OK;
 638     void *chunk = NULL, *hchunk = NULL;
 639     void *tptr =NULL;
 640     int ret = 0;
 641     if (takeLock) {
 642         rv = systemDatabase_->getXCheckpointMutex();
 643         if (OK != rv)
 644         {
 645             printError(ErrSysInternal, "Unable to get database mutex");
 646             return ErrSysInternal;
 647         }
 648     }
 649     rv = removeIndexCatalogTables(name, chunk, hchunk, tptr);
 650     if (OK != rv)
 651     {
 652         if (takeLock) systemDatabase_->releaseCheckpointMutex();
 653         return rv;
 654     }
 655
 656     CINDEX *iptr = (CINDEX*)tptr;
 657     rv = removeIndexChunks(chunk, hchunk, iptr->indexType_);
 658     if (OK != rv)
 659     {
 660         if (takeLock) systemDatabase_->releaseCheckpointMutex();
 661         return rv;
 662     }
 663     if (takeLock) systemDatabase_->releaseCheckpointMutex();
 664
 665     printDebug(DM_Database, "Dropped index %s",name);
 666     logFinest(Conf::logger, "Deleted Index %s", name);
 667     return OK;
 668 }
 669
 670 void DatabaseManagerImpl::printTreeIndexNodeInfo(char *name, bool flag)
 671 {
 672     CatalogTableINDEX cIndex(systemDatabase_);
 673     DbRetVal rv = OK;
 674     void *chunk = NULL, *hchunk = NULL;
 675     void *tptr =NULL;
 676     rv = cIndex.get(name, chunk, hchunk, tptr);
 677     if (OK != rv) return;
 678     IndexType iType = CatalogTableINDEX::getType(tptr);
 679     if (treeIndex != iType)
 680     {
 681        printf("%s is not a tree index\n ");
 682        return;
 683     }
 684     Chunk *ch = (Chunk*) chunk;
 685     if(flag){  if(hchunk)((TreeNode*) hchunk)->displayAll(); }
 686     else {
 687         int offset = CatalogTableINDEX::getOffsetOfFirstField(tptr);
 688         //if(typeInt != offset) { printf("%s is not on Integer Type Field. To see info Index should be on integer type field. \n "); return;}
 689         if(hchunk) ((TreeNode*) hchunk)->displayAll(offset);
 690     }
 691 }
 692 DbRetVal DatabaseManagerImpl::printIndexDebugInfo(char *name)
 693 {
 694     CatalogTableINDEX cIndex(systemDatabase_);
 695     DbRetVal rv = OK;
 696     void *chunk = NULL, *hchunk = NULL;
 697     void *tptr =NULL;
 698     rv = cIndex.get(name, chunk, hchunk, tptr);
 699     if (OK != rv) return rv;
 700     IndexType iType = CatalogTableINDEX::getType(tptr);
 701     if(hashIndex == iType) {
 702     }
 703     else if (treeIndex == iType) {
 704         CINDEX *iptr = (CINDEX*)tptr;
 705         TreeNode *start = (TreeNode*) iptr->hashNodeChunk_;
 706         start->displayAll(0);
 707     }
 708     else if (trieIndex == iType) {
 709         ChunkIterator citer = CatalogTableINDEX::getIterator(tptr);
 710         TrieNode* start = (TrieNode*)citer.nextElement();
 711         if(start) TrieIndex::displayAll(start);
 712     }
 713     else
 714         printf("Unknown Index\n");
 715
 716
 717 }
 718
 719 DbRetVal DatabaseManagerImpl::printIndexInfo(char *name)
 720 {
 721     CatalogTableINDEX cIndex(systemDatabase_);
 722     DbRetVal rv = OK;
 723     void *chunk = NULL, *hchunk = NULL;
 724     void *tptr =NULL;
 725     rv = cIndex.get(name, chunk, hchunk, tptr);
 726     if (OK != rv) return rv;
 727     printf("<IndexName> %s </IndexName>\n", name);
 728     printf("<Unique> %d </Unique>\n", CatalogTableINDEX::getUnique(tptr));
 729     IndexType iType = CatalogTableINDEX::getType(tptr);
 730     if(hashIndex == iType)
 731         printf("<Type> Hash Index </Type>\n");
 732     else if (treeIndex == iType)
 733         printf("<Type> Tree Index </Type>\n");
 734     else if (trieIndex == iType)
 735         printf("<Type> Trie Index </Type>\n");
 736     else
 737         printf("<Type> Unknown Index </Type>\n");
 738
 739     Chunk *ch = (Chunk*) chunk;
 740     printf("<HashBucket>\n");
 741     printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
 742     printf("  <TotalBuckets> %d </TotalBuckets> \n", CatalogTableINDEX::getNoOfBuckets(tptr));
 743     printf("</HashBucket>\n");
 744     printf("<IndexNodes>\n");
 745     if(hashIndex == iType){
 746         ch = (Chunk*) hchunk;
 747         printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
 748         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
 749     } else if (treeIndex == iType) {
 750         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
 751         if(hchunk)
 752             printf("  <TotalElements> %lld </TotalElements>\n",((TreeNode*) hchunk)->getTotalElements());
 753         else
 754             printf("  <TotalElements> 0 </TotalElements>\n");
 755     } else if (trieIndex == iType)
 756     {
 757         printf(" <TrieNodes> \n");
 758         printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
 759         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
 760         printf(" </TrieNodes> \n <TrieValues>\n");
 761         ch = (Chunk*) hchunk;
 762         printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
 763         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
 764         printf(" </TrieValues>\n");
 765     } else
 766     {
 767         printf("Unknown Index type\n");
 768     }
 769     printf("<IndexNodes>\n");
 770     return OK;
 771 }