src/server/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 HashIndex::computeHashBucket(DataType type, void *key, int noOfBuckets)
  49
  50 {
  51     switch(type)
  52     {
  53         case typeInt:
  54         {
  55             int val = *(int*)key;
  56             return val % noOfBuckets;
  57             break;
  58         }
  59         case typeLong:
  60         {
  61             long val = *(long*)key;
  62             return val % noOfBuckets;
  63             break;
  64         }
  65         case typeULong:
  66         {
  67             unsigned long val = *(unsigned long*)key;
  68             return val % noOfBuckets;
  69             break;
  70         }
  71         case typeLongLong:
  72         {
  73             long long val = *(long long*)key;
  74             return val % noOfBuckets;
  75             break;
  76         }
  77         case typeShort:
  78         {
  79             short val = *(short*)key;
  80             return val % noOfBuckets;
  81             break;
  82         }
  83         case typeByteInt:
  84         {
  85             ByteInt val = *(ByteInt*)key;
  86             return val % noOfBuckets;
  87             break;
  88         }
  89         case typeDate:
  90         {
  91             int val = *(int*)key;
  92             return val % noOfBuckets;
  93             break;
  94         }
  95         case typeTime:
  96         {
  97             int val = *(int*)key;
  98             return val % noOfBuckets;
  99             break;
 100         }
 101         case typeTimeStamp:
 102         {
 103             TimeStamp val = *(TimeStamp*)key;
 104             //TODO return val % noOfBuckets;
 105             break;
 106         }
 107         case typeDouble:
 108         {
 109             //TODO
 110             break;
 111         }
 112         case typeFloat:
 113         {
 114             //TODO
 115             break;
 116         }
 117         case typeDecimal:
 118         {
 119             //TODO::for porting
 120         }
 121         case typeString:
 122         {
 123             unsigned int val = hashString((char*)key);
 124             return val % noOfBuckets;
 125         }
 126         case typeBinary:
 127         {
 128             //TODO
 129         }
 130         default:
 131         {
 132             break;
 133         }
 134     }
 135     return -1;
 136 }
 137
 138 //TODO::composite keys are not supported currently
 139 DbRetVal HashIndex::insert(TableImpl *tbl, Transaction *tr, void *indexPtr, IndexInfo *indInfo, void *tuple, bool undoFlag)
 140 {
 141     SingleFieldHashIndexInfo *info = (SingleFieldHashIndexInfo*) indInfo;
 142     INDEX *iptr = (INDEX*)indexPtr;
 143     DbRetVal rc = OK;
 144     DataType type = info->type;
 145     char *name = info->fldName;
 146     int offset  = info->offset;
 147     int noOfBuckets = info->noOfBuckets;
 148     int length = info->length;
 149
 150     printDebug(DM_HashIndex, "Inserting hash index node for field %s", name);
 151     ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
 152     Bucket* buckets = (Bucket*)citer.nextElement();
 153     void *keyPtr =(void*)((char*)tuple + offset);
 154
 155     int bucketNo = computeHashBucket(type,
 156                         keyPtr, noOfBuckets);
 157     printDebug(DM_HashIndex, "HashIndex insert bucketno %d", bucketNo);
 158     Bucket *bucket =  &(buckets[bucketNo]);
 159
 160     int ret = bucket->mutex_.getLock(tbl->db_->procSlot);
 161     if (ret != 0)
 162     {
 163         printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",bucketNo);
 164         return ErrLockTimeOut;
 165     }
 166     HashIndexNode *head = (HashIndexNode*) bucket->bucketList_;
 167     if (info->isUnique)
 168     {
 169         BucketList list(head);
 170         BucketIter iter = list.getIterator();
 171         HashIndexNode *node;
 172         void *bucketTuple;
 173         printDebug(DM_HashIndex, "HashIndex insert Checking for unique");
 174         while((node = iter.next()) != NULL)
 175         {
 176             bucketTuple = node->ptrToTuple_;
 177             if (AllDataType::compareVal((void*)((char*)bucketTuple +offset),
 178                    (void*)((char*)tuple +offset), OpEquals,type, length))
 179             {
 180                 printError(ErrUnique, "Unique key violation");
 181                 bucket->mutex_.releaseLock(tbl->db_->procSlot);
 182                 return ErrUnique;
 183             }
 184         }
 185     }
 186
 187     printDebug(DM_HashIndex, "HashIndex insert into bucket list");
 188     if (!head)
 189     {
 190         printDebug(DM_HashIndex, "HashIndex insert head is empty");
 191         DbRetVal rv = OK;
 192         HashIndexNode *firstNode= (HashIndexNode*)(((Chunk*)iptr->hashNodeChunk_)->allocate(tbl->db_, &rv));
 193         if (firstNode == NULL)
 194         {
 195             bucket->mutex_.releaseLock(tbl->db_->procSlot);
 196             return rv;
 197         }
 198         firstNode->ptrToKey_ = keyPtr;
 199         firstNode->ptrToTuple_ = tuple;
 200         firstNode->next_ = NULL;
 201         bucket->bucketList_ = (HashIndexNode*)firstNode;
 202         printDebug(DM_HashIndex, "HashIndex insert new node %x in empty bucket", bucket->bucketList_);
 203     }
 204     else
 205     {
 206         BucketList list(head);
 207         rc = list.insert((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr, tuple);
 208         if (rc !=OK) {
 209             bucket->mutex_.releaseLock(tbl->db_->procSlot);
 210             printf("PRABA::bucket insert failed here with rc %d\n", rc);
 211             return rc;
 212         }
 213
 214     }
 215     if (undoFlag) {
 216
 217          rc = tr->appendLogicalUndoLog(tbl->sysDB_, InsertHashIndexOperation,
 218                                  tuple, sizeof(void*), indexPtr);
 219         if (rc !=OK)
 220         {
 221             BucketList list(head);
 222             rc = list.remove((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr);
 223             if (rc !=OK) printError(ErrSysFatal, "double failure on undo log insert followed by hash bucket list remove\n");
 224         }
 225     }
 226
 227     bucket->mutex_.releaseLock(tbl->db_->procSlot);
 228     return rc;
 229 }
 230
 231
 232 //TODO::composite keys are not supported currently
 233 DbRetVal HashIndex::remove(TableImpl *tbl, Transaction *tr, void *indexPtr, IndexInfo *indInfo, void *tuple, bool undoFlag)
 234 {
 235     INDEX *iptr = (INDEX*)indexPtr;
 236
 237     SingleFieldHashIndexInfo *info = (SingleFieldHashIndexInfo*) indInfo;
 238     DataType type = info->type;
 239     char *name = info->fldName;
 240     int offset  = info->offset;
 241     int noOfBuckets = info->noOfBuckets;
 242
 243     printDebug(DM_HashIndex, "Removing hash index node for field %s", name);
 244     ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
 245     Bucket* buckets = (Bucket*)citer.nextElement();
 246     void *keyPtr =(void*)((char*)tuple + offset);
 247
 248     int bucket = HashIndex::computeHashBucket(type, keyPtr, noOfBuckets);
 249
 250     Bucket *bucket1 = &buckets[bucket];
 251
 252     int ret = bucket1->mutex_.getLock(tbl->db_->procSlot);
 253     if (ret != 0)
 254     {
 255         printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",bucket);
 256         return ErrLockTimeOut;
 257     }
 258     HashIndexNode *head = (HashIndexNode*) bucket1->bucketList_;
 259
 260     if (!head) { printError(ErrNotExists, "Hash index does not exist:should never happen\n"); return ErrNotExists; }
 261     BucketList list(head);
 262     printDebug(DM_HashIndex, "Removing hash index node from head %x", head);
 263
 264     DbRetVal rc = list.remove((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr);
 265     if (SplCase == rc)
 266     {
 267        printDebug(DM_HashIndex, "Removing hash index node from head with only none node");
 268        printError(ErrWarning, "Removing hash index node from head with only none node");
 269        bucket1->bucketList_ = 0;
 270        rc = OK;
 271     }
 272     if (undoFlag) {
 273         rc =tr->appendLogicalUndoLog(tbl->sysDB_, DeleteHashIndexOperation,
 274                                 tuple, sizeof(void*), indexPtr);
 275         if (rc !=OK)
 276         {
 277         //TODO::add it back to the bucketlist
 278         }
 279     }
 280     bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 281     return rc;
 282 }
 283
 284 //TODO::composite keys are not supported currently
 285 DbRetVal HashIndex::update(TableImpl *tbl, Transaction *tr, void *indexPtr, IndexInfo *indInfo, void *tuple, bool undoFlag)
 286 {
 287     INDEX *iptr = (INDEX*)indexPtr;
 288
 289     SingleFieldHashIndexInfo *info = (SingleFieldHashIndexInfo*) indInfo;
 290     DataType type = info->type;
 291     char *name = info->fldName;
 292     int offset  = info->offset;
 293     int noOfBuckets = info->noOfBuckets;
 294
 295     printDebug(DM_HashIndex, "Updating hash index node for field %s", name);
 296
 297     //check whether the index key is updated or not
 298     //if it is not updated return from here
 299     void *keyPtr =(void*)((char*)tuple + offset);
 300     //Iterate through the bind list and check
 301     FieldIterator fldIter = tbl->fldList_.getIterator();
 302     void *newKey = NULL;
 303     while (fldIter.hasElement())
 304     {
 305         FieldDef def = fldIter.nextElement();
 306         if (0 == strcmp(def.fldName_, name))
 307         {
 308             if (NULL == def.bindVal_)
 309                 return OK;
 310             bool result = AllDataType::compareVal(keyPtr, def.bindVal_,
 311                                 OpEquals, def.type_, def.length_);
 312             if (result) return OK; else newKey = def.bindVal_;
 313         }
 314     }
 315     printDebug(DM_HashIndex, "Updating hash index node: Key value is updated");
 316
 317     if (newKey == NULL)
 318     {
 319         printError(ErrSysInternal,"Internal Error:: newKey is Null");
 320         return ErrSysInternal;
 321     }
 322     ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
 323
 324     Bucket* buckets = (Bucket*)citer.nextElement();
 325
 326     //remove the node whose key is updated
 327     int bucketNo = computeHashBucket(type,
 328                         keyPtr, noOfBuckets);
 329     printDebug(DM_HashIndex, "Updating hash index node: Bucket for old value is %d", bucketNo);
 330     Bucket *bucket = &buckets[bucketNo];
 331
 332     //it may run into deadlock, when two threads updates tuples which falls in
 333     //same buckets.So take both the mutex one after another, which will reduce the
 334     //deadlock window.
 335     int ret = bucket->mutex_.getLock(tbl->db_->procSlot);
 336     if (ret != 0)
 337     {
 338         printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",bucketNo);
 339         return ErrLockTimeOut;
 340     }
 341     //insert node for the updated key value
 342     int newBucketNo = computeHashBucket(type,
 343                         newKey, noOfBuckets);
 344     printDebug(DM_HashIndex, "Updating hash index node: Bucket for new value is %d", newBucketNo);
 345
 346     Bucket *bucket1 = &buckets[newBucketNo];
 347     bucket1->mutex_.getLock(tbl->db_->procSlot);
 348     if (ret != 0)
 349     {
 350         printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",newBucketNo);
 351         return ErrLockTimeOut;
 352     }
 353
 354     HashIndexNode *head1 = (HashIndexNode*) bucket->bucketList_;
 355     if (head1)
 356     {
 357         BucketList list1(head1);
 358         printDebug(DM_HashIndex, "Updating hash index node: Removing node from list with head %x", head1);
 359         list1.remove((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr);
 360     }
 361     else
 362     {
 363         printError(ErrSysInternal,"Update: Bucket list is null");
 364         return ErrSysInternal;
 365     }
 366     DbRetVal rc = OK;
 367     if (undoFlag) {
 368          rc = tr->appendLogicalUndoLog(tbl->sysDB_, DeleteHashIndexOperation,
 369                                 tuple, sizeof(void*), indexPtr);
 370         if (rc !=OK)
 371         {
 372             //TODO::add it back to the bucket list
 373             bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 374             bucket->mutex_.releaseLock(tbl->db_->procSlot);
 375             return rc;
 376         }
 377     }
 378     HashIndexNode *head2 = (HashIndexNode*) bucket1->bucketList_;
 379     //Note:: the tuple will be in the same address location
 380     //so not changing the keyptr and tuple during append
 381     //only bucket where this node resides will only change
 382     //if the index key is updated.
 383     if (!head2)
 384     {
 385         DbRetVal rv = OK;
 386         HashIndexNode *firstNode= (HashIndexNode*)(((Chunk*)iptr->hashNodeChunk_)->allocate(tbl->db_, &rv));
 387         if (firstNode == NULL)
 388         {
 389             printError(rv, "Error in allocating hash node");
 390             bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 391             bucket->mutex_.releaseLock(tbl->db_->procSlot);
 392             return rv;
 393         }
 394         firstNode->ptrToKey_ = keyPtr;
 395         firstNode->ptrToTuple_ = tuple;
 396         firstNode->next_ = NULL;
 397         bucket1->bucketList_ = (HashIndexNode*)firstNode;
 398         printDebug(DM_HashIndex, "Updating hash index node: Adding new node %x:Head is empty", firstNode);
 399     }
 400     else
 401     {
 402         BucketList list2(head2);
 403         printDebug(DM_HashIndex, "Updating hash index node: Adding node to list with head %x", head2);
 404         list2.insert((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr, tuple);
 405     }
 406     if (undoFlag) {
 407
 408         rc = tr->appendLogicalUndoLog(tbl->sysDB_, InsertHashIndexOperation,
 409                                 tuple, sizeof(void*), indexPtr);
 410         if (rc !=OK)
 411         {
 412             //TODO::revert back the changes:delete and add the node + remove the logical undo log
 413             //of the  DeleteHashIndexOperation
 414         }
 415     }
 416     bucket1->mutex_.releaseLock(tbl->db_->procSlot);
 417     bucket->mutex_.releaseLock(tbl->db_->procSlot);
 418     return rc;
 419 }