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<TableConfig.h>
  29 #include<Process.h>
  30
  31 DbRetVal DatabaseManagerImpl::renameIndex(const char *oldName,const char *newName)
  32 {
  33     void *chunk = NULL;
  34     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
  35     if (OK != rv) {
  36         printError(ErrSysInternal, "Unable to get database mutex for rename table");
  37         return ErrSysInternal;
  38     }
  39     CatalogTableTABLE cTable(systemDatabase_);
  40     rv = cTable.renameIndex(oldName,newName);
  41     if (OK != rv) {
  42         printError(ErrSysInternal, "Unable to rename table");
  43         systemDatabase_->releaseCheckpointMutex();
  44         return ErrSysInternal;
  45     }
  46     systemDatabase_->releaseCheckpointMutex();
  47     return OK;
  48 }
  49
  50 DbRetVal DatabaseManagerImpl::createIndex(const char *indName, IndexInitInfo *info)
  51 {
  52     DbRetVal rv = OK;
  53     if (!info->isUnique && info->isPrimary)
  54     {
  55         printError(ErrBadCall, "Primary key cannot be non unique\n");
  56         return ErrBadCall;
  57     }
  58     if (!Util::isIdentifier((char*)indName)) {
  59         printError(ErrBadArg, "Invalid character for index name");
  60         return ErrBadArg;
  61     }
  62
  63     if (info->indType == hashIndex)
  64     {
  65         //Assumes info is of type HashIndexInitInfo
  66         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
  67         rv = createHashIndex(indName, info->tableName, info->list, hInfo->bucketSize,
  68                              info->isUnique, info->isPrimary);
  69     }
  70     else if (info->indType == treeIndex)
  71     {
  72         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
  73         rv = createTreeIndex(indName, info->tableName, info->list,
  74                           hInfo->bucketSize, info->isUnique, info->isPrimary);
  75     }
  76     else if (info->indType == trieIndex)
  77     {
  78         HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
  79         rv = createTrieIndex(indName, info->tableName, info->list,
  80                           info->isUnique, info->isPrimary);
  81
  82     }else {
  83         printError(ErrBadCall, "Index type not supported\n");
  84         return ErrBadCall;
  85     }
  86     return rv;
  87 }
  88
  89
  90 //-1 -> Table does not exists
  91 //-2 -> Field does not exists
  92 //-3 -> bucketSize is not valid
  93 DbRetVal DatabaseManagerImpl::createHashIndex(const char *indName, const char *tblName,
  94                       FieldNameList &fldList, int bucketSize, bool isUnique, bool isPrimary)
  95 {
  96     //validate the bucket size
  97     if (bucketSize < 100 || bucketSize > 200000)
  98     {
  99         printError(ErrBadRange, "Index Bucket size %d not in range 100-200000",
 100                                  bucketSize);
 101         return ErrBadRange;
 102     }
 103     int totFlds = fldList.size();
 104     if (totFlds == 0)
 105     {
 106         printError(ErrBadCall, "No Field name specified");
 107         return ErrBadCall;
 108     }
 109     void *tptr =NULL;
 110     void *chunk = NULL;
 111     void *vcchunk = NULL;
 112     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 113     if (OK != rv)
 114     {
 115         printError(ErrSysInternal, "Unable to get database mutex");
 116         return ErrSysInternal;
 117     }
 118
 119     //check whether table exists
 120     CatalogTableTABLE cTable(systemDatabase_);
 121     cTable.getChunkAndTblPtr(tblName, chunk, tptr, vcchunk);
 122     if (NULL == tptr)
 123     {
 124         systemDatabase_->releaseCheckpointMutex();
 125         printError(ErrNotExists, "Table does not exist %s", tblName);
 126         return ErrNotExists;
 127     }
 128
 129     //check whether field exists
 130     char **fptr = new char* [totFlds];
 131     CatalogTableFIELD cField(systemDatabase_);
 132     rv = cField.getFieldPtrs(fldList, tptr, fptr);
 133     if (OK != rv)
 134     {
 135         delete[] fptr;
 136         systemDatabase_->releaseCheckpointMutex();
 137         //TODO::check test cases of dbapi/Index, they give wrong results
 138         //if (rv == ErrBadCall) {
 139         ////    if (isPrimary) printError(ErrBadCall, "Field can have NULL values");
 140         //} else {
 141         //printError(ErrNotExists, "Field does not exist");
 142         //}
 143         //return ErrBadCall;
 144         if (rv != ErrBadCall) {
 145             printError(ErrNotExists, "Field does not exist");
 146             return ErrNotExists;
 147         }
 148     }
 149     for (int i=0; i <totFlds; i++)
 150     {
 151         CFIELD* fInfo = (CFIELD*)fptr[i];
 152         if (fInfo->type_ == typeFloat || fInfo->type_ == typeDouble || fInfo->type_ == typeTimeStamp)
 153         {
 154             printError(ErrBadArg, "HashIndex cannot be created for float or double or timestamp type");
 155             delete[] fptr;
 156             systemDatabase_->releaseCheckpointMutex();
 157             return ErrBadArg;
 158         }
 159         if (!fInfo->isNull_ && isPrimary )
 160         {
 161             printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
 162             delete[] fptr;
 163             systemDatabase_->releaseCheckpointMutex();
 164             return ErrBadArg;
 165         }
 166          if(isPrimary){fInfo->isPrimary_=true;fInfo->isUnique_=true;}
 167          if(isUnique){fInfo->isUnique_=true;}
 168     }
 169     //create chunk to store the meta data of the index created
 170     //for latches and bucket pointers
 171     printDebug(DM_HashIndex, "Creating chunk for storing hash buckets of size %d\n",
 172                              bucketSize * sizeof(Bucket));
 173     Chunk* chunkInfo = createUserChunk(bucketSize * sizeof(Bucket));
 174     if (NULL == chunkInfo)
 175     {
 176         delete[] fptr;
 177         systemDatabase_->releaseCheckpointMutex();
 178         printError(ErrSysInternal, "Unable to create chunk");
 179         return ErrSysInternal;
 180     }
 181     chunkInfo->setChunkName(indName);
 182     //create memory for holding the bucket pointers
 183     void *buckets = chunkInfo->allocate(db_, &rv);
 184     if (NULL == buckets)
 185     {
 186         delete[] fptr;
 187         deleteUserChunk(chunkInfo);
 188         systemDatabase_->releaseCheckpointMutex();
 189         printError(rv, "Unable to allocate memory for bucket");
 190         return rv;
 191     }
 192     Bucket *buck = (Bucket*) buckets;
 193     initHashBuckets(buck, bucketSize);
 194
 195     //create chunk to store the hash index nodes
 196     Chunk* hChunk = createUserChunk(sizeof(IndexNode));
 197     if (NULL == hChunk)
 198     {
 199         delete[] fptr;
 200         deleteUserChunk(chunkInfo);
 201         systemDatabase_->releaseCheckpointMutex();
 202         printError(ErrSysInternal, "Unable to create chunk for storing hash index nodes");
 203         return ErrSysInternal;
 204     }
 205
 206      hChunk->setChunkName(indName);
 207     //add row to INDEX
 208     void *tupleptr = NULL;
 209     CatalogTableINDEX cIndex(systemDatabase_);
 210     rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
 211                         chunkInfo, bucketSize, hChunk, tupleptr);
 212     if (OK != rv)
 213     {
 214         delete[] fptr;
 215         deleteUserChunk(hChunk);
 216         deleteUserChunk(chunkInfo);
 217         systemDatabase_->releaseCheckpointMutex();
 218         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
 219         return ErrSysInternal;
 220     }
 221     //add rows to INDEXFIELD
 222     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 223     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
 224
 225     if (OK != rv)
 226     {
 227         delete[] fptr;
 228         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
 229         deleteUserChunk(hChunk);
 230         deleteUserChunk(chunkInfo);
 231         systemDatabase_->releaseCheckpointMutex();
 232         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
 233         return ErrSysInternal;
 234     }
 235     delete[] fptr;
 236     systemDatabase_->releaseCheckpointMutex();
 237
 238     //TODO:: Take table lock
 239
 240     // Following code is written by Kishor Amballi
 241     TableImpl *tbl = (TableImpl *) openTable(tblName);
 242     if (NULL == tbl) {
 243         printError(ErrSysInternal, "Unable to open table %s", tblName);
 244         return ErrSysInternal;
 245     }
 246     if (! tbl->numTuples()) {
 247         printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x", indName, tblName, buckets);
 248         logFinest(Conf::logger, "Creating HashIndex %s on %s with bucket size %d", indName, tblName, buckets);
 249         closeTable(tbl);
 250         return OK;
 251     }
 252     HashIndexInfo *indxInfo = NULL;
 253     int i = 0;
 254     for (i = 0; i < tbl->numIndexes_; i++) {
 255         if(((HashIndexInfo *)tbl->idxInfo[i])->indexPtr == tupleptr) {
 256             indxInfo = (HashIndexInfo *) tbl->idxInfo[i];
 257             break;
 258         }
 259     }
 260     void *recPtr = NULL;
 261     ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
 262     tbl->setLoading(true);
 263     while ((recPtr = chIter.nextElement()) != NULL) {
 264         rv = tbl->insertIndexNode(*tbl->trans, tupleptr, indxInfo, recPtr);
 265         if (rv == ErrUnique) {
 266             closeTable(tbl);
 267             dropIndex(indName);
 268             return rv;
 269         }
 270     }
 271     closeTable(tbl);
 272     printDebug(DM_Database, "Creating Hash Index Name:%s tblname:%s buckets:%x", indName, tblName, buckets);
 273     logFinest(Conf::logger, "Creating HashIndex %s on %s with bucket size %d", indName, tblName, buckets);
 274     return OK;
 275 }
 276
 277
 278 DbRetVal DatabaseManagerImpl::createTreeIndex(const char *indName, const char *tblName,
 279                       FieldNameList &fldList, int nodeSize, bool isUnique, bool isPrimary)
 280 {
 281     if (nodeSize < 20 || nodeSize > 20000)
 282     {
 283         printError(ErrBadRange,"Tree Index Node size %d not in range 20-20000",
 284                                  nodeSize);
 285         return ErrBadRange;
 286     }
 287     int totFlds = fldList.size();
 288     if (totFlds == 0) {
 289         printError(ErrBadCall, "No Field name specified");
 290         return ErrBadCall;
 291     }else if (totFlds != 1) {
 292         printError(ErrBadCall, "Composite index not supported for Tree");
 293         return ErrBadCall;
 294     }
 295     void *tptr =NULL;
 296     void *chunk = NULL;
 297     void *vcchunk = NULL;
 298     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 299     if (OK != rv)
 300     {
 301         printError(ErrSysInternal, "Unable to get database mutex");
 302         return ErrSysInternal;
 303     }
 304     //check whether table exists
 305
 306     CatalogTableTABLE cTable(systemDatabase_);
 307     cTable.getChunkAndTblPtr(tblName, chunk, tptr, vcchunk);
 308     if (NULL == tptr)
 309     {
 310         systemDatabase_->releaseCheckpointMutex();
 311         printError(ErrNotExists, "Table does not exist %s", tblName);
 312         return ErrNotExists;
 313     }
 314     char **fptr = new char* [totFlds];
 315     CatalogTableFIELD cField(systemDatabase_);
 316     rv = cField.getFieldPtrs(fldList, tptr, fptr);
 317     if (OK != rv)
 318     {
 319         delete[] fptr;
 320         systemDatabase_->releaseCheckpointMutex();
 321         if (rv != ErrBadCall) {
 322             printError(ErrNotExists, "Field does not exist");
 323             return ErrNotExists;
 324         }
 325     }
 326     for (int i=0; i <totFlds; i++)
 327     {
 328         CFIELD* fInfo = (CFIELD*)fptr[i];
 329         if (!fInfo->isNull_ && isPrimary )
 330         {
 331             printError(ErrBadArg, "Primary Index cannot be created on field without NOTNULL constraint");
 332             delete[] fptr;
 333             systemDatabase_->releaseCheckpointMutex();
 334             return ErrBadArg;
 335         }
 336         if (fInfo->type_ == typeVarchar)
 337         {
 338             printError(ErrBadArg, "Tree Index not supported for varchar type. Use char data type instead.");
 339             delete[] fptr;
 340            systemDatabase_->releaseCheckpointMutex();
 341             return ErrBadArg;
 342         }
 343
 344     }
 345     int chunkSize = sizeof(TreeNode)+(nodeSize * sizeof(void*));
 346     printDebug(DM_HashIndex, "Creating chunk for storing tree nodes of size %d\n", chunkSize);
 347
 348     Chunk* chunkInfo = createUserChunk(chunkSize);
 349     if (NULL == chunkInfo)
 350     {
 351         delete[] fptr;
 352         systemDatabase_->releaseCheckpointMutex();
 353         printError(ErrSysInternal, "Unable to create chunk");
 354         return ErrSysInternal;
 355     }
 356
 357
 358     void *tupleptr = NULL;
 359
 360     CatalogTableINDEX cIndex(systemDatabase_);
 361     rv = cIndex.insert(indName, tptr, fldList.size(), isUnique,
 362                         chunkInfo, nodeSize, NULL, tupleptr);
 363     if (OK != rv)
 364     {
 365         delete[] fptr;
 366         deleteUserChunk(chunkInfo);
 367         systemDatabase_->releaseCheckpointMutex();
 368         printError(ErrSysInternal, "Catalog table updation failed in INDEX table");
 369         return ErrSysInternal;
 370     }
 371     CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 372     rv = cIndexField.insert(fldList, tupleptr, tptr, fptr);
 373
 374     if (OK != rv)
 375     {
 376         delete[] fptr;
 377         void *hChunk = NULL;
 378         cIndex.remove(indName, (void *&)chunkInfo, (void *&)hChunk, (void *&)tupleptr);
 379         deleteUserChunk(chunkInfo);
 380         systemDatabase_->releaseCheckpointMutex();
 381         printError(ErrSysInternal, "Catalog table updation failed in INDEXFIELD table");
 382         return ErrSysInternal;
 383     }
 384     delete[] fptr;
 385     rv = createIndexNodeForRecords(tblName, tupleptr, chunk);
 386     if (rv != OK)
 387     {
 388         dropIndex(indName);
 389         systemDatabase_->releaseCheckpointMutex();
 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 DbRetVal DatabaseManagerImpl::removeIndexChunks(void* chunk, void* hchunk, IndexType iType)
 598 {
 599     DbRetVal rv = deleteUserChunk((Chunk*)chunk);
 600     if (OK != rv)
 601     {
 602         printError(ErrSysInternal, "Unable to delete the index chunk");
 603         return ErrSysInternal;
 604     }
 605     //delete the index hash node chunk
 606     if (iType == hashIndex || iType == trieIndex) {
 607         rv = deleteUserChunk((Chunk*)hchunk);
 608         if (OK != rv)
 609         {
 610             printError(ErrSysInternal, "Unable to delete the index hash node chunk");
 611             return ErrSysInternal;
 612         }
 613     }
 614     return OK;
 615 }
 616
 617 void DatabaseManagerImpl::initHashBuckets(Bucket *buck, int bucketSize)
 618 {
 619     os::memset((void*)buck, 0, bucketSize * sizeof(Bucket));
 620     char mutName[IDENTIFIER_LENGTH];
 621     for (int i=0; i < bucketSize ; i++)
 622     {
 623         sprintf(mutName, "BKT:%d",i);
 624         buck[i].mutex_.init(mutName);
 625     }
 626     return;
 627 }
 628
 629 DbRetVal DatabaseManagerImpl::dropIndex(const char *name)
 630 {
 631     return dropIndexInt(name, true);
 632 }
 633
 634 DbRetVal DatabaseManagerImpl::dropIndexInt(const char *name, bool takeLock)
 635 {
 636     DbRetVal rv = OK;
 637     void *chunk = NULL, *hchunk = NULL;
 638     void *tptr =NULL;
 639     int ret = 0;
 640     if (takeLock) {
 641         rv = systemDatabase_->getXCheckpointMutex();
 642         if (OK != rv)
 643         {
 644             printError(ErrSysInternal, "Unable to get database mutex");
 645             return ErrSysInternal;
 646         }
 647     }
 648     rv = removeIndexCatalogTables(name, chunk, hchunk, tptr);
 649     if (OK != rv)
 650     {
 651         if (takeLock) systemDatabase_->releaseCheckpointMutex();
 652         return rv;
 653     }
 654
 655     CINDEX *iptr = (CINDEX*)tptr;
 656     rv = removeIndexChunks(chunk, hchunk, iptr->indexType_);
 657     if (OK != rv)
 658     {
 659         if (takeLock) systemDatabase_->releaseCheckpointMutex();
 660         return rv;
 661     }
 662     if (takeLock) systemDatabase_->releaseCheckpointMutex();
 663
 664     printDebug(DM_Database, "Dropped index %s",name);
 665     logFinest(Conf::logger, "Deleted Index %s", name);
 666     return OK;
 667 }
 668
 669 void DatabaseManagerImpl::printTreeIndexNodeInfo(char *name, bool flag)
 670 {
 671     CatalogTableINDEX cIndex(systemDatabase_);
 672     DbRetVal rv = OK;
 673     void *chunk = NULL, *hchunk = NULL;
 674     void *tptr =NULL;
 675     rv = cIndex.get(name, chunk, hchunk, tptr);
 676     if (OK != rv) return;
 677     IndexType iType = CatalogTableINDEX::getType(tptr);
 678     if (treeIndex != iType)
 679     {
 680        printf("%s is not a tree index\n ");
 681        return;
 682     }
 683     Chunk *ch = (Chunk*) chunk;
 684     if(flag){  if(hchunk)((TreeNode*) hchunk)->displayAll(); }
 685     else {
 686         int offset = CatalogTableINDEX::getOffsetOfFirstField(tptr);
 687         //if(typeInt != offset) { printf("%s is not on Integer Type Field. To see info Index should be on integer type field. \n "); return;}
 688         if(hchunk) ((TreeNode*) hchunk)->displayAll(offset);
 689     }
 690 }
 691 DbRetVal DatabaseManagerImpl::printIndexDebugInfo(char *name)
 692 {
 693     CatalogTableINDEX cIndex(systemDatabase_);
 694     DbRetVal rv = OK;
 695     void *chunk = NULL, *hchunk = NULL;
 696     void *tptr =NULL;
 697     rv = cIndex.get(name, chunk, hchunk, tptr);
 698     if (OK != rv) return rv;
 699     IndexType iType = CatalogTableINDEX::getType(tptr);
 700     if(hashIndex == iType) {
 701     }
 702     else if (treeIndex == iType) {
 703         CINDEX *iptr = (CINDEX*)tptr;
 704         TreeNode *start = (TreeNode*) iptr->hashNodeChunk_;
 705         start->displayAll(0);
 706     }
 707     else if (trieIndex == iType) {
 708         ChunkIterator citer = CatalogTableINDEX::getIterator(tptr);
 709         TrieNode* start = (TrieNode*)citer.nextElement();
 710         if(start) TrieIndex::displayAll(start);
 711     }
 712     else
 713         printf("Unknown Index\n");
 714
 715
 716 }
 717 DbRetVal DatabaseManagerImpl::printIndexInfo(char *name)
 718 {
 719     CatalogTableINDEX cIndex(systemDatabase_);
 720     DbRetVal rv = OK;
 721     void *chunk = NULL, *hchunk = NULL;
 722     void *tptr =NULL;
 723     rv = cIndex.get(name, chunk, hchunk, tptr);
 724     if (OK != rv) return rv;
 725     printf("<IndexName> %s </IndexName>\n", name);
 726     printf("<Unique> %d </Unique>\n", CatalogTableINDEX::getUnique(tptr));
 727     IndexType iType = CatalogTableINDEX::getType(tptr);
 728     if(hashIndex == iType)
 729         printf("<Type> Hash Index </Type>\n");
 730     else if (treeIndex == iType)
 731         printf("<Type> Tree Index </Type>\n");
 732     else if (trieIndex == iType)
 733         printf("<Type> Trie Index </Type>\n");
 734     else
 735         printf("<Type> Unknown Index </Type>\n");
 736
 737     Chunk *ch = (Chunk*) chunk;
 738     printf("<HashBucket>\n");
 739     printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
 740     printf("  <TotalBuckets> %d </TotalBuckets> \n", CatalogTableINDEX::getNoOfBuckets(tptr));
 741     printf("</HashBucket>\n");
 742     printf("<IndexNodes>\n");
 743     if(hashIndex == iType){
 744         ch = (Chunk*) hchunk;
 745         printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
 746         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
 747     } else if (treeIndex == iType) {
 748         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
 749         if(hchunk)
 750             printf("  <TotalElements> %lld </TotalElements>\n",((TreeNode*) hchunk)->getTotalElements());
 751         else
 752             printf("  <TotalElements> 0 </TotalElements>\n");
 753     } else if (trieIndex == iType)
 754     {
 755         printf(" <TrieNodes> \n");
 756         printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
 757         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
 758         printf(" </TrieNodes> \n <TrieValues>\n");
 759         ch = (Chunk*) hchunk;
 760         printf("  <TotalPages> %d </TotalPages>\n", ch->totalPages());
 761         printf("  <TotalNodes> %d </TotalNodes>\n", ch->getTotalDataNodes());
 762         printf(" </TrieValues>\n");
 763     } else
 764     {
 765         printf("Unknown Index type\n");
 766     }
 767     printf("<IndexNodes>\n");
 768     return OK;
 769 }