src/storage/HashIndex.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<Index.h>
  17 #include<CatalogTables.h>
  18 #include<Lock.h>
  19 #include<Debug.h>
  20 #include<Table.h>
  21 #include<TableImpl.h>
  22 #include<Predicate.h>
  23 #include<PredicateImpl.h>
  24
  25 /* Defines `hashpjw' function by P.J. Weinberger
  26    [see Aho/Sethi/Ullman, COMPILERS: Principles, Techniques and Tools,
  27    */
  28
  29 unsigned int hashString(char *strVal)
  30 {
  31     unsigned int hval, g;
  32     hval = 0;
  33     char *str =strVal;
  34     while (*str != '\0')
  35     {
  36         hval <<= 4;
  37         hval += (unsigned int) *str++;
  38         g = hval & ((unsigned int) 0xf << (32 - 4));
  39         if (g != 0)
  40             {
  41                 hval ^= g >> (32 - 8);
  42                 hval ^= g;
  43             }
  44     }
  45     return hval;
  46 }
  47
  48 unsigned int hashBinary(char *strVal, int length)
  49 {
  50     unsigned int hval, g;
  51     hval = 0;
  52     char *str =strVal;
  53     int iter = 0;
  54     while (iter != length)
  55     {
  56         hval <<= 4;
  57         hval += (unsigned int) *str++;
  58         g = hval & ((unsigned int) 0xf << (32 - 4));
  59         if (g != 0)
  60             {
  61                 hval ^= g >> (32 - 8);
  62                 hval ^= g;
  63             }
  64         iter++;
  65     }
  66     return hval;
  67 }
  68
  69 unsigned int HashIndex::computeHashBucket(DataType type, void *key, int noOfBuckets, int length)
  70
  71 {
  72     switch(type)
  73     {
  74         case typeInt:
  75         {
  76             int val = *(int*)key;
  77             return val % noOfBuckets;
  78             break;
  79         }
  80         case typeLong:
  81         {
  82             long val = *(long*)key;
  83             return val % noOfBuckets;
  84             break;
  85         }
  86         case typeULong:
  87         {
  88             unsigned long val = *(unsigned long*)key;
  89             return val % noOfBuckets;
  90             break;
  91         }
  92         case typeLongLong:
  93         {
  94             long long val = *(long long*)key;
  95             return val % noOfBuckets;
  96             break;
  97         }
  98         case typeShort:
  99         {
 100             short val = *(short*)key;
 101             return val % noOfBuckets;
 102             break;
 103         }
 104         case typeByteInt:
 105         {
 106             ByteInt val = *(ByteInt*)key;
 107             return val % noOfBuckets;
 108             break;
 109         }
 110         case typeDate:
 111         {
 112             int val = *(int*)key;
 113             return val % noOfBuckets;
 114             break;
 115         }
 116         case typeTime:
 117         {
 118             int val = *(int*)key;
 119             return val % noOfBuckets;
 120             break;
 121         }
 122         case typeTimeStamp:
 123         {
 124             TimeStamp val = *(TimeStamp*)key;
 125             //TODO return val % noOfBuckets;
 126             break;
 127         }
 128         case typeDouble:
 129         {
 130             //TODO
 131             break;
 132         }
 133         case typeFloat:
 134         {
 135             //TODO
 136             break;
 137         }
 138         case typeDecimal:
 139         {
 140             //TODO::for porting
 141         }
 142         case typeString:
 143         {
 144             unsigned int val = hashString((char*)key);
 145             return val % noOfBuckets;
 146         }
 147         case typeComposite:
 148         case typeBinary:
 149         {
 150             unsigned int val = hashBinary((char*)key, length);
 151             return val % noOfBuckets;
 152         }
 153         default:
 154         {
 155             break;
 156         }
 157     }
 158     return -1;
 159 }
 160
 161 DbRetVal HashIndex::insert(TableImpl *tbl, Transaction *tr, void *indexPtr, IndexInfo *indInfo, void *tuple, bool undoFlag)
 162 {
 163     HashIndexInfo *info = (HashIndexInfo*) indInfo;
 164     CINDEX *iptr = (CINDEX*)indexPtr;
 165     DbRetVal rc = OK;
 166     int noOfBuckets = info->noOfBuckets;
 167     int offset = info->fldOffset;
 168     DataType type = info->type;
 169     char *keyBuffer = (char*) malloc(info->compLength);
 170     void *keyStartBuffer = keyBuffer, *keyPtr;
 171     FieldIterator iter = info->idxFldList.getIterator();
 172     while(iter.hasElement())
 173     {
 174         FieldDef def = iter.nextElement();
 175         keyPtr = (char *)tuple + def.offset_;
 176         AllDataType::copyVal(keyBuffer, keyPtr, def.type_, def.length_);
 177         keyBuffer = keyBuffer + AllDataType::size(def.type_, def.length_);
 178     }
 179
 180     printDebug(DM_HashIndex, "Inserting hash index node for  %s", iptr->indName_);
 181     ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
 182     Bucket* buckets = (Bucket*)citer.nextElement();
 183     keyPtr =(void*)((char*)tuple + offset);
 184     int bucketNo = 0;
 185     if (type == typeComposite)
 186         bucketNo = computeHashBucket(type, keyStartBuffer, noOfBuckets, info->compLength);
 187     else
 188         bucketNo = computeHashBucket(type, keyPtr, noOfBuckets, info->compLength);
 189     printDebug(DM_HashIndex, "HashIndex insert bucketno %d", bucketNo);
 190     Bucket *bucket =  &(buckets[bucketNo]);
 191
 192     int ret = bucket->mutex_.getLock(tbl->db_->procSlot);
 193     if (ret != 0)
 194     {
 195         printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",bucketNo);
 196         return ErrLockTimeOut;
 197     }
 198     HashIndexNode *head = (HashIndexNode*) bucket->bucketList_;
 199     if (info->isUnique)
 200     {
 201         BucketList list(head);
 202         BucketIter iter = list.getIterator();
 203         HashIndexNode *node;
 204         void *bucketTuple;
 205         printDebug(DM_HashIndex, "HashIndex insert Checking for unique");
 206         bool res = false;
 207
 208         while((node = iter.next()) != NULL)
 209         {
 210             bucketTuple = node->ptrToTuple_;
 211             if (type == typeComposite) {
 212                 FieldIterator fldIter = info->idxFldList.getIterator();
 213                 int i = 0;
 214                 while (fldIter.hasElement()) {
 215                     FieldDef def = fldIter.nextElement();
 216                     res = AllDataType::compareVal((char *)bucketTuple + def.offset_, (char *)tuple + def.offset_, OpEquals, def.type_, def.length_);
 217                     if (!res) break;
 218                 }
 219             }
 220             else res = AllDataType::compareVal((void*)((char*)bucketTuple +offset), (void*)((char*)tuple +offset), OpEquals,type, info->compLength);
 221             if (res)
 222             {
 223                 printError(ErrUnique, "Unique key violation");
 224                 bucket->mutex_.releaseLock(tbl->db_->procSlot);
 225                 return ErrUnique;
 226             }
 227         }
 228     }
 229
 230     printDebug(DM_HashIndex, "HashIndex insert into bucket list");
 231     if (!head)
 232     {
 233         printDebug(DM_HashIndex, "HashIndex insert head is empty");
 234         DbRetVal rv = OK;
 235         HashIndexNode *firstNode= (HashIndexNode*)(((Chunk*)iptr->hashNodeChunk_)->allocate(tbl->db_, &rv));
 236         if (firstNode == NULL)
 237         {
 238             bucket->mutex_.releaseLock(tbl->db_->procSlot);
 239             return rv;
 240         }
 241         firstNode->ptrToKey_ = keyPtr;
 242         firstNode->ptrToTuple_ = tuple;
 243         firstNode->next_ = NULL;
 244         bucket->bucketList_ = (HashIndexNode*)firstNode;
 245         printDebug(DM_HashIndex, "HashIndex insert new node %x in empty bucket", bucket->bucketList_);
 246     }
 247     else
 248     {
 249         BucketList list(head);
 250         rc = list.insert((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr, tuple);
 251         if (rc !=OK) {
 252             bucket->mutex_.releaseLock(tbl->db_->procSlot);
 253             return rc;
 254         }
 255
 256     }
 257     if (undoFlag) {
 258
 259          rc = tr->appendLogicalUndoLog(tbl->sysDB_, InsertHashIndexOperation,
 260                                  tuple, sizeof(void*), indexPtr);
 261         if (rc !=OK)
 262         {
 263             BucketList list(head);
 264             rc = list.remove((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr);
 265             if (rc !=OK) printError(ErrSysFatal, "double failure on undo log insert followed by hash bucket list remove\n");
 266         }
 267     }
 268
 269     bucket->mutex_.releaseLock(tbl->db_->procSlot);
 270     return rc;
 271 }
 272
 273
 274 DbRetVal HashIndex::remove(TableImpl *tbl, Transaction *tr, void *indexPtr, IndexInfo *indInfo, void *tuple, bool undoFlag)
 275 {
 276     CINDEX *iptr = (CINDEX*)indexPtr;
 277
 278     HashIndexInfo *info = (HashIndexInfo*) indInfo;
 279     DataType type = info->type;
 280     int offset  = info->fldOffset;
 281     int noOfBuckets = info->noOfBuckets;
 282
 283     ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
 284     Bucket* buckets = (Bucket*)citer.nextElement();
 285
 286     char *keyBuffer = (char*) malloc(info->compLength);
 287     void *keyStartBuffer = keyBuffer, *keyPtr;
 288     FieldIterator iter = info->idxFldList.getIterator();
 289     while(iter.hasElement())
 290     {
 291         FieldDef def = iter.nextElement();
 292         keyPtr = (char *)tuple + def.offset_;
 293         AllDataType::copyVal(keyBuffer, keyPtr, def.type_, def.length_);
 294         keyBuffer = keyBuffer + AllDataType::size(def.type_, def.length_);
 295     }
 296
 297     keyPtr =(void*)((char*)tuple + offset);
 298     int bucket = 0;
 299     if (type == typeComposite)
 300         bucket = HashIndex::computeHashBucket(type, keyStartBuffer, noOfBuckets, info->compLength);
 301     else bucket = HashIndex::computeHashBucket(type, keyPtr, noOfBuckets, info->compLength);
 302
 303     Bucket *bucket1 = &buckets[bucket];
 304
 305     int ret = bucket1->mutex_.getLock(tbl->db_->procSlot);
 306     if (ret != 0)
 307     {
 308         printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",bucket);
 309         return ErrLockTimeOut;
 310     }
 311     HashIndexNode *head = (HashIndexNode*) bucket1->bucketList_;
 312
 313     if (!head) { printError(ErrNotExists, "Hash index does not exist:should never happen\n");
 314        bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 315        return ErrNotExists;
 316     }
 317     BucketList list(head);
 318     printDebug(DM_HashIndex, "Removing hash index node from head %x", head);
 319
 320     DbRetVal rc = list.remove((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr);
 321     if (SplCase == rc)
 322     {
 323        printDebug(DM_HashIndex, "Removing hash index node from head ");
 324        bucket1->bucketList_ = list.getBucketListHead();
 325        rc = OK;
 326     }
 327     if (undoFlag) {
 328         rc =tr->appendLogicalUndoLog(tbl->sysDB_, DeleteHashIndexOperation,
 329                                 tuple, sizeof(void*), indexPtr);
 330         if (rc !=OK)
 331         {
 332         //TODO::add it back to the bucketlist
 333         }
 334     }
 335     bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 336
 337     return rc;
 338 }
 339
 340 DbRetVal HashIndex::update(TableImpl *tbl, Transaction *tr, void *indexPtr, IndexInfo *indInfo, void *tuple, bool undoFlag)
 341 {
 342     CINDEX *iptr = (CINDEX*)indexPtr;
 343
 344     HashIndexInfo *info = (HashIndexInfo*) indInfo;
 345     DataType type = info->type;
 346     int offset  = info->fldOffset;
 347     int noOfBuckets = info->noOfBuckets;
 348
 349     //check whether the index key is updated or not
 350     //if it is not updated return from here
 351     void *keyPtr =(void*)((char*)tuple + offset);
 352     char *kPtr= (char*)keyPtr;
 353
 354     //creating old key value buffer for composite primary keys
 355     char *oldKeyBuffer = (char*) malloc(info->compLength);
 356     void *oldKeyStartBuffer = oldKeyBuffer;
 357     FieldIterator iter = info->idxFldList.getIterator();
 358     while(iter.hasElement()) {
 359         FieldDef def = iter.nextElement();
 360         keyPtr = (char *)tuple + def.offset_;
 361         AllDataType::copyVal(oldKeyBuffer, keyPtr, def.type_, def.length_);
 362         oldKeyBuffer = oldKeyBuffer + AllDataType::size(def.type_, def.length_);
 363     }
 364
 365     keyPtr = (void *) kPtr;
 366     //Iterate through the bind list and check
 367     FieldIterator idxFldIter = info->idxFldList.getIterator();
 368     char *keyBindBuffer ;
 369     if(type==typeBinary) keyBindBuffer = (char*) malloc(2 * info->compLength);
 370     else keyBindBuffer = (char*) malloc(info->compLength);
 371     void *keyStartBuffer = (void*) keyBindBuffer;
 372     bool keyUpdated = false;
 373
 374     while (idxFldIter.hasElement()) {
 375         FieldDef idef = idxFldIter.nextElement();
 376         FieldIterator fldIter = tbl->fldList_.getIterator();
 377         while (fldIter.hasElement()) {
 378             FieldDef def = fldIter.nextElement();
 379             if (0 == strcmp(def.fldName_, idef.fldName_)) {
 380                 if (NULL != def.bindVal_) {
 381                     if(type==typeBinary) {
 382                         AllDataType::copyVal(keyBindBuffer, def.bindVal_,
 383                                                    def.type_, 2*def.length_);
 384                         keyStartBuffer=calloc(1,info->compLength);
 385                         AllDataType::convertToBinary(keyStartBuffer, keyBindBuffer, typeString, info->compLength);
 386                         free(keyBindBuffer);
 387                     } else {
 388                         AllDataType::copyVal(keyBindBuffer, def.bindVal_,
 389                                                    def.type_, def.length_);
 390                         keyBindBuffer = keyBindBuffer + AllDataType::size(def.type_, def.length_);
 391                     }
 392                 } else {
 393                      AllDataType::copyVal(keyBindBuffer, (char *) tuple + def.offset_, def.type_, def.length_);
 394                      keyBindBuffer = keyBindBuffer + AllDataType::size(def.type_, def.length_);
 395                 }
 396                 keyUpdated = true;
 397                 break;
 398             }
 399         }
 400     }
 401     if (!keyUpdated) {
 402         //printf("PRABA::key not updated\n");
 403         free(keyStartBuffer);
 404         return OK;
 405     }
 406     //printf("PRABA::it is wrong coming here\n");
 407     bool result = false;
 408     if (type == typeComposite)
 409         result = AllDataType::compareVal(oldKeyStartBuffer, keyStartBuffer,
 410                                 OpEquals, info->type, info->compLength);
 411     else result = AllDataType::compareVal(keyPtr, keyStartBuffer,
 412                                 OpEquals, info->type, info->compLength);
 413     if (result) return OK;
 414     printDebug(DM_HashIndex, "Updating hash index node: Key value is updated");
 415
 416     ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
 417
 418     Bucket* buckets = (Bucket*)citer.nextElement();
 419
 420     //remove the node whose key is updated
 421     int bucketNo = 0;
 422     if (type == typeComposite)
 423         bucketNo = computeHashBucket(type, oldKeyStartBuffer, noOfBuckets, info->compLength);
 424     else bucketNo = computeHashBucket(type, keyPtr, noOfBuckets, info->compLength);
 425     printDebug(DM_HashIndex, "Updating hash index node: Bucket for old value is %d", bucketNo);
 426     Bucket *bucket = &buckets[bucketNo];
 427
 428     //it may run into deadlock, when two threads updates tuples which falls in
 429     //same buckets.So take both the mutex one after another, which will reduce the
 430     //deadlock window.
 431     int ret = bucket->mutex_.getLock(tbl->db_->procSlot);
 432     if (ret != 0)
 433     {
 434         printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",bucketNo);
 435         return ErrLockTimeOut;
 436     }
 437     //insert node for the updated key value
 438     int newBucketNo = computeHashBucket(type,
 439                         keyStartBuffer, noOfBuckets, info->compLength);
 440     printDebug(DM_HashIndex, "Updating hash index node: Bucket for new value is %d", newBucketNo);
 441
 442     Bucket *bucket1 = &buckets[newBucketNo];
 443     bucket1->mutex_.getLock(tbl->db_->procSlot);
 444     if (ret != 0)
 445     {
 446         printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",newBucketNo);
 447         return ErrLockTimeOut;
 448     }
 449
 450     HashIndexNode *head1 = (HashIndexNode*) bucket->bucketList_;
 451     if (head1)
 452     {
 453         BucketList list1(head1);
 454         printDebug(DM_HashIndex, "Updating hash index node: Removing node from list with head %x", head1);
 455         list1.remove((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr);
 456         bucket->bucketList_=list1.getBucketListHead();
 457     }
 458     else
 459     {
 460         printError(ErrSysInternal,"Update: Bucket list is null");
 461         bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 462         bucket->mutex_.releaseLock(tbl->db_->procSlot);
 463         return ErrSysInternal;
 464     }
 465     DbRetVal rc = OK;
 466     if (undoFlag) {
 467          rc = tr->appendLogicalUndoLog(tbl->sysDB_, DeleteHashIndexOperation,
 468                                 tuple, sizeof(void*), indexPtr);
 469         if (rc !=OK)
 470         {
 471             //TODO::add it back to the bucket list
 472             bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 473             bucket->mutex_.releaseLock(tbl->db_->procSlot);
 474             return rc;
 475         }
 476     }
 477     HashIndexNode *head2 = (HashIndexNode*) bucket1->bucketList_;
 478     //Note:: the tuple will be in the same address location
 479     //so not changing the keyptr and tuple during append
 480     //only bucket where this node resides will only change
 481     //if the index key is updated.
 482     if (!head2)
 483     {
 484         DbRetVal rv = OK;
 485         HashIndexNode *firstNode= (HashIndexNode*)(((Chunk*)iptr->hashNodeChunk_)->allocate(tbl->db_, &rv));
 486         if (firstNode == NULL)
 487         {
 488             printError(rv, "Error in allocating hash node");
 489             bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 490             bucket->mutex_.releaseLock(tbl->db_->procSlot);
 491             return rv;
 492         }
 493         firstNode->ptrToKey_ = keyPtr;
 494         firstNode->ptrToTuple_ = tuple;
 495         firstNode->next_ = NULL;
 496         bucket1->bucketList_ = (HashIndexNode*)firstNode;
 497         printDebug(DM_HashIndex, "Updating hash index node: Adding new node %x:Head is empty", firstNode);
 498     }
 499     else
 500     {
 501         BucketList list2(head2);
 502         printDebug(DM_HashIndex, "Updating hash index node: Adding node to list with head %x", head2);
 503         list2.insert((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr, tuple);
 504     }
 505     if (undoFlag) {
 506
 507         rc = tr->appendLogicalUndoLog(tbl->sysDB_, InsertHashIndexOperation,
 508                                 tuple, sizeof(void*), indexPtr);
 509         if (rc !=OK)
 510         {
 511             //TODO::revert back the changes:delete and add the node + remove the logical undo log
 512             //of the  DeleteHashIndexOperation
 513         }
 514     }
 515     bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 516     bucket->mutex_.releaseLock(tbl->db_->procSlot);
 517
 518     return rc;
 519 }