src/storage/TableImpl.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 #include<Index.h>
  25 #include<Config.h>
  26 #include<AggTableImpl.h> //for AggType
  27
  28 void Table::getFieldNameAlone(char *fname, char *name) {
  29     bool dotFound= false;
  30     char *fullname = fname;
  31     while(*fullname != '\0')
  32     {
  33         if (*fullname == '.') { dotFound = true;  break; }
  34         fullname++;
  35     }
  36     if (dotFound) strcpy(name, ++fullname); else strcpy(name, fname);
  37
  38 }
  39 void Table::getTableNameAlone(char *fname, char *name) {
  40     strcpy(name, fname);
  41     char *start = name;
  42     bool dotFound = false;
  43     while(*name != '\0')
  44     {
  45         if (*name == '.') { *name='\0';  dotFound = true; break; }
  46         name++;
  47     }
  48     if (!dotFound) strcpy(start, "");
  49     return;
  50 }
  51
  52 DbRetVal TableImpl::bindFld(const char *name, void *val)
  53 {
  54     if (name[0] == '*' ) return OK;
  55     //set it in the field list
  56     char fieldName[IDENTIFIER_LENGTH];
  57     getFieldNameAlone((char*)name, fieldName);
  58     DbRetVal rv = fldList_.updateBindVal(fieldName, val);
  59     if (OK != rv) {
  60         printError(ErrNotExists, "Field %s does not exist", fieldName);
  61         return  rv;
  62     }
  63     return OK;
  64 }
  65
  66 bool TableImpl::isFldNull(const char *name){
  67     if (name[0] == '*') return false;
  68     char fieldName[IDENTIFIER_LENGTH];
  69     getFieldNameAlone((char*)name, fieldName);
  70     int colpos = fldList_.getFieldPosition(fieldName);
  71     if (-1 == colpos)
  72     {
  73         printError(ErrNotExists, "Field %s does not exist", name);
  74         return false;
  75     }
  76
  77     return isFldNull(colpos);
  78 }
  79
  80 int TableImpl::getFldPos(char *name)
  81 {
  82     return fldList_.getFieldPosition(name);
  83 }
  84 bool TableImpl::isFldNull(int colpos)
  85 {
  86     if (!curTuple_) return false;
  87     if (colpos <1 || colpos > numFlds_) return false;
  88     if (isIntUsedForNULL) {
  89         int nullVal = *(int*)((char*)curTuple_ + (length_ - 4));
  90         if (BITSET(nullVal, colpos)) return true;
  91     }
  92     else {
  93         char *nullOffset = (char*)curTuple_ - os::align(numFlds_);
  94         if (nullOffset[colpos-1]) return true;
  95     }
  96     return false;
  97 }
  98 void TableImpl::resetNullinfo()
  99 {
 100     if (isIntUsedForNULL) {
 101         iNullInfo =0;
 102     }
 103     else {
 104         int i=0;
 105         while(i < numFlds_) { cNullInfo[0] = 0;}
 106     }
 107 }
 108 DbRetVal TableImpl::markFldNull(char const* name)
 109 {
 110     DbRetVal rv = OK;
 111     int colpos = fldList_.getFieldPosition(name);
 112     if (-1 == colpos)
 113     {
 114         printError(ErrNotExists, "Field %s does not exist", name);
 115         return ErrNotExists;
 116     }
 117     rv =  markFldNull(colpos);
 118     return rv;
 119 }
 120
 121 DbRetVal TableImpl::markFldNull(int fldpos)
 122 {
 123     if (fldpos <1 || fldpos > numFlds_) return ErrBadArg;
 124     bool isBitSet = false;
 125     if (isIntUsedForNULL) {
 126         if (!BITSET(iNotNullInfo, fldpos)) {
 127             SETBIT(iNullInfo, fldpos);
 128             isBitSet = true;
 129         }
 130         else {
 131             printError(ErrNullViolation, "NOT NULL constraint violation");
 132             return ErrNullViolation;
 133         }
 134     }
 135     else {
 136         if (!BITSET(iNotNullInfo, fldpos)) cNullInfo[fldpos-1] = 1;
 137         else {
 138             printError(ErrNullViolation, "NOT NULL constraint violation");
 139             return ErrNullViolation;
 140         }
 141     }
 142     return OK;
 143 }
 144
 145 void TableImpl::clearFldNull(const char *name)
 146 {
 147     int colpos = fldList_.getFieldPosition(name);
 148     if (-1 == colpos)
 149     {
 150         printError(ErrNotExists, "Field %s does not exist", name);
 151         return;
 152     }
 153
 154     clearFldNull(colpos);
 155 }
 156
 157 void TableImpl::clearFldNull(int colpos)
 158 {
 159     if (colpos <1 || colpos > numFlds_) return;
 160     if (isIntUsedForNULL) {
 161         CLEARBIT(iNullInfo, colpos);
 162     }
 163     else
 164         cNullInfo[colpos-1] = 0;
 165     return;
 166 }
 167
 168 bool TableImpl::hasIndex(char* fName)
 169 {
 170    if (NULL == indexPtr_) return false;
 171    for (int i =0; i < numIndexes_; i++)
 172    {
 173       HashIndexInfo* info = (HashIndexInfo*) idxInfo[i];
 174       FieldIterator iter = info->idxFldList.getIterator();
 175       if(iter.hasElement())
 176       {
 177         FieldDef *def = iter.nextElement();
 178         if(strcmp(def->fldName_, fName) == 0)
 179             if(!iter.hasElement())//neglet if it is composite index
 180                 return true;
 181       }
 182    }
 183    return false;
 184 }
 185
 186 IndexType TableImpl::getIndexType(char *fName, int *pos)
 187 {
 188    if (NULL == indexPtr_) return unknownIndex;
 189    for (int i =0; i < numIndexes_; i++)
 190    {
 191       HashIndexInfo* info = (HashIndexInfo*) idxInfo[i];
 192       FieldIterator iter = info->idxFldList.getIterator();
 193       if(iter.hasElement())
 194       {
 195         FieldDef *def = iter.nextElement();
 196         if(strcmp(def->fldName_, fName) == 0)
 197             if(!iter.hasElement()) {//neglet if it is composite index
 198                 *(int*)pos = i;
 199                 return info->indType;
 200             }
 201       }
 202    }
 203    *(int*)pos = -1;
 204    return unknownIndex;
 205 }
 206 void TableImpl::addPredicate(char *fName, ComparisionOp op, void *buf)
 207 {
 208     char fieldName[IDENTIFIER_LENGTH];
 209     Table::getFieldNameAlone(fName, fieldName);
 210     PredicateImpl *pred = (PredicateImpl*) pred_;
 211     PredicateImpl *newPred = new PredicateImpl();
 212     newPred->setTerm(fName, op, buf);
 213     if (NULL == pred) { pred_ = newPred; return; }
 214     if (pred->isSingleTerm())
 215     {
 216        bool res = pred->appendIfSameFld(fName, op, buf);
 217        if(res) {
 218            delete newPred;
 219            return;
 220        }
 221     }
 222     PredicateImpl *bothPred = new PredicateImpl();
 223     bothPred->setTerm(pred, OpAnd, newPred);
 224     pred_ = bothPred;
 225 }
 226
 227 DbRetVal TableImpl::optimize()
 228 {
 229     //table ptr is set in predicate because it needs to access the
 230     //type and length to evaluate
 231     if( NULL != pred_)
 232     {
 233         PredicateImpl *pred = (PredicateImpl*) pred_;
 234         pred->setTable(this);
 235         pred->setProjectionList(NULL);
 236         pred->setOffsetAndType();
 237     }
 238     return createPlan();
 239 }
 240
 241 DbRetVal TableImpl::execute()
 242 {
 243     if (NULL != iter)
 244     {
 245          //printError(ErrAlready,"Scan already open:Close and re execute");
 246          return ErrAlready;
 247     }
 248     DbRetVal ret = OK;
 249     ret = optimize();
 250     if (OK != ret)
 251     {
 252         printError(ErrSysInternal,"Unable to create the plan");
 253         return ErrSysInternal;
 254     }
 255     if (useIndex_ >= 0)
 256         iter = new TupleIterator(pred_, scanType_, idxInfo[useIndex_], chunkPtr_, sysDB_->procSlot,isBetween,isPointLook);
 257     else if (scanType_ == fullTableScan)
 258         iter = new TupleIterator(pred_, scanType_, NULL, chunkPtr_, sysDB_->procSlot,isBetween,isPointLook);
 259     else
 260     {
 261         printError(ErrSysFatal,"Unable to create tuple iterator");//should never happen
 262         return ErrSysFatal;
 263     }
 264     ret = iter->open();
 265     if (OK != ret)
 266     {
 267         printError(ret,"Unable to open the iterator");
 268         return ret;
 269     }
 270     return OK;
 271 }
 272
 273
 274 DbRetVal TableImpl::createPlan()
 275 {
 276     if (isPlanCreated) {
 277         //will do early return here. plan is generated only when setPredicate is called.
 278         if (scanType_ == unknownScan) return ErrSysFatal; //this should never happen
 279         else return OK;
 280     }
 281     isBetween=false;
 282     isPointLook = false;
 283     useIndex_ = -1;
 284
 285     FieldIterator fIter = fldList_.getIterator();
 286     FieldDef *def = NULL;
 287     while ((def = fIter.nextElement())!= NULL) {
 288         if (NULL != def->bindVal_) bindList_.append(def);
 289     }
 290     numBindFlds_ = bindList_.size();
 291     bindListArray_ = (void **) malloc(numBindFlds_ * sizeof (void *));
 292     void *elem = NULL;
 293     int i = 0;
 294     ListIterator it = bindList_.getIterator();
 295     while ((elem = it.nextElement()) != NULL) bindListArray_[i++] = elem;
 296
 297     //if there are no predicates then go for full scan
 298     //if there are no indexes then go for full scan
 299     if (NULL == pred_ || NULL == indexPtr_)
 300     {
 301         scanType_ = fullTableScan;
 302         isPlanCreated = true;
 303         return OK;
 304     }
 305     if (NULL != indexPtr_)
 306     {
 307        PredicateImpl *pred = (PredicateImpl*)pred_;
 308        printDebug(DM_Predicate, "predicate does not involve NOT , OR operator");
 309        if (!pred->isNotOrInvolved())
 310        {
 311            printDebug(DM_Predicate, "predicate does not involve NOT , OR operator");
 312           for (int i =0; i < numIndexes_; i++)
 313           {
 314               HashIndexInfo* info = (HashIndexInfo*) idxInfo[i];
 315               FieldIterator iter = info->idxFldList.getIterator();
 316               while(iter.hasElement())
 317               {
 318                 FieldDef *def = iter.nextElement();
 319                 if (pred->pointLookupInvolved(def->fldName_))
 320                 {
 321                   printDebug(DM_Predicate, "point lookup involved for field %s",def->fldName_);
 322                   if(hashIndex == info->indType) scanType_ = hashIndexScan;
 323                   else scanType_ = treeIndexScan;
 324                   isPlanCreated = true;
 325                   isPointLook = true;
 326                   useIndex_ = i;
 327                 }
 328                 else if (pred->isBetweenInvolved(def->fldName_))
 329                 {
 330                     if (treeIndex == info->indType)
 331                     {
 332                      scanType_ = treeIndexScan;
 333                      isPlanCreated = true;
 334                      useIndex_ = i;
 335                      isBetween=true;
 336                      break; //no composite index for tree index
 337                     }
 338                 }
 339                 else if (pred->rangeQueryInvolved(def->fldName_))
 340                 {
 341                   printDebug(DM_Predicate, "range lookup involved for field %s",def->fldName_);
 342                   if (treeIndex == info->indType)
 343                   {
 344                      scanType_ = treeIndexScan;
 345                      isPlanCreated = true;
 346                      useIndex_ = i;
 347                      break; //no composite index for tree index
 348                   }
 349                 }else {
 350                     useIndex_ = -1;
 351                     break;
 352                 }
 353              }//while iter.hasElement()
 354              if (useIndex_ != -1) return OK;
 355            }//for
 356         }
 357     }
 358     scanType_ = fullTableScan;
 359     isPlanCreated = true;
 360     return OK;
 361 }
 362
 363 void* TableImpl::fetch()
 364 {
 365     fetchNoBind();
 366     if (NULL == curTuple_) return curTuple_;
 367     copyValuesToBindBuffer(curTuple_);
 368     return curTuple_;
 369 }
 370 void* TableImpl::fetch(DbRetVal &rv)
 371 {
 372     fetchNoBind(rv);
 373     if (NULL == curTuple_) return curTuple_;
 374     copyValuesToBindBuffer(curTuple_);
 375     return curTuple_;
 376 }
 377
 378 void* TableImpl::fetchNoBind()
 379 {
 380     if (NULL == iter)
 381     {
 382         printError(ErrNotOpen,"Scan not open or Scan is closed\n");
 383         return NULL;
 384     }
 385     void *prevTuple = curTuple_;
 386     curTuple_ = iter->next();
 387     if (NULL == curTuple_)
 388     {
 389         return NULL;
 390     }
 391     DbRetVal lockRet = OK;
 392     if ((*trans)->isoLevel_ == READ_COMMITTED)
 393     {
 394         //if iso level is read committed, operation duration lock is sufficent
 395         //so release it here itself.
 396         int tries = 5;
 397         struct timeval timeout;
 398         timeout.tv_sec = Conf::config.getMutexSecs();
 399         timeout.tv_usec = Conf::config.getMutexUSecs();
 400
 401         bool status = false;
 402         while(true) {
 403             lockRet = lMgr_->isExclusiveLocked( curTuple_, trans, status);
 404             if (OK != lockRet)
 405             {
 406                 printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 407                 curTuple_ = prevTuple;
 408                 return NULL;
 409             }
 410             if (!status) break;
 411             tries--;
 412             if (tries == 0) break;
 413             os::select(0, 0, 0, 0, &timeout);
 414
 415         }
 416         if (tries == 0)
 417         {
 418             printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 419             curTuple_ = prevTuple;
 420             return NULL;
 421         }
 422     }
 423     else if ((*trans)->isoLevel_ == READ_REPEATABLE) {
 424         lockRet = lMgr_->getSharedLock(curTuple_, trans);
 425         if (OK != lockRet)
 426         {
 427             printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 428             curTuple_ = prevTuple;
 429             return NULL;
 430         }
 431
 432     }
 433     return curTuple_;
 434 }
 435
 436 void* TableImpl::fetchNoBind(DbRetVal &rv)
 437 {
 438     rv = OK;
 439     if (NULL == iter)
 440     {
 441         printError(ErrNotOpen,"Scan not open or Scan is closed\n");
 442         rv = ErrNotOpen;
 443         return NULL;
 444     }
 445     void *prevTuple = curTuple_;
 446     curTuple_ = iter->next();
 447     if (NULL == curTuple_)
 448     {
 449         return NULL;
 450     }
 451     DbRetVal lockRet = OK;
 452     if ((*trans)->isoLevel_ == READ_REPEATABLE) {
 453         lockRet = lMgr_->getSharedLock(curTuple_, trans);
 454         if (OK != lockRet)
 455         {
 456             printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 457             rv = ErrLockTimeOut;
 458             curTuple_ = prevTuple;
 459             return NULL;
 460         }
 461
 462     }
 463     else if ((*trans)->isoLevel_ == READ_COMMITTED)
 464     {
 465         //if iso level is read committed, operation duration lock is sufficent
 466         //so release it here itself.
 467         int tries = 5;
 468         struct timeval timeout;
 469         timeout.tv_sec = Conf::config.getMutexSecs();
 470         timeout.tv_usec = Conf::config.getMutexUSecs();
 471
 472         bool status = false;
 473         while(true) {
 474             lockRet = lMgr_->isExclusiveLocked( curTuple_, trans, status);
 475             if (OK != lockRet)
 476             {
 477                 printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 478                 curTuple_ = prevTuple;
 479                 rv = ErrLockTimeOut;
 480                 return NULL;
 481             }
 482             if (!status) break;
 483             tries--;
 484             if (tries == 0) break;
 485             os::select(0, 0, 0, 0, &timeout);
 486
 487         }
 488         if (tries == 0)
 489         {
 490             printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 491             curTuple_ = prevTuple;
 492             rv = ErrLockTimeOut;
 493             return NULL;
 494         }
 495     }
 496     return curTuple_;
 497 }
 498 DbRetVal TableImpl::fetchAgg(const char * fldName, AggType aType, void *buf)
 499 {
 500    FieldInfo *info = new FieldInfo();
 501    DbRetVal rv = getFieldInfo(fldName, info);
 502    if (OK != rv) return rv;
 503    bool res= false;
 504    if (AGG_MIN == aType || AGG_MAX == aType) {
 505       int pos =0;
 506       IndexType iType = getIndexType((char*)fldName, &pos);
 507       if(treeIndex == iType && pos >=0) {
 508           if (AGG_MIN == aType) {
 509               HashIndexInfo* hInfo = (HashIndexInfo*) idxInfo[pos];
 510               CINDEX *iptr = (CINDEX*) hInfo->indexPtr;
 511               TreeIter *iter = new TreeIter((TreeNode*)iptr->hashNodeChunk_);
 512               char *tuple = (char*) iter->getFirstElement();
 513               if (tuple != NULL) {
 514                   AllDataType::copyVal(buf,(void*)(tuple+info->offset),
 515                                    info->type, info->length);
 516                   delete iter;
 517                   return OK;
 518               }
 519               delete iter;
 520           }
 521           else if (AGG_MAX == aType) {
 522               HashIndexInfo* hInfo = (HashIndexInfo*) idxInfo[pos];
 523               CINDEX *iptr = (CINDEX*) hInfo->indexPtr;
 524               TreeIter *iter = new TreeIter((TreeNode*)iptr->hashNodeChunk_);
 525               char *tuple = (char*) iter->getLastElement();
 526               if (tuple != NULL) {
 527                   AllDataType::copyVal(buf,(void*)(tuple+info->offset),
 528                                    info->type, info->length);
 529                   delete iter;
 530                   return OK;
 531               }
 532               delete iter;
 533           }
 534       }
 535    }
 536
 537    DataType type = info->type;
 538    int length = info->length;
 539    int offset = info->offset;
 540
 541    if (NULL == pred_ && typeInt == type)
 542    { //perf opt
 543      ChunkIterator cIter = ((Chunk*)chunkPtr_)->getIterator();
 544      char *tuple =(char*)cIter.nextElement();
 545      if (NULL == tuple) return OK;
 546      int count =1;
 547      AllDataType::copyVal(buf, (void*) (tuple+offset), type, length);
 548      while(1) {
 549        tuple = (char*)cIter.nextElement();
 550        if (NULL == tuple) break;
 551        switch(aType) {
 552            case AGG_MIN:
 553            {
 554                if (*(int*)buf >= *((int*)(tuple+offset)))
 555                    *(int*)buf = *((int*)(tuple+offset));
 556                break;
 557            }
 558            case AGG_MAX:
 559            {
 560                if (*(int*)buf <= *((int*)(tuple+offset)))
 561                    *(int*)buf = *((int*)(tuple+offset));
 562                break;
 563            }
 564            case AGG_SUM:
 565            {
 566                *(int*)buf = *(int*)buf + *((int*)(tuple+offset));
 567                break;
 568            }
 569            case AGG_AVG:
 570            {
 571                *(int*)buf = *(int*)buf + *((int*)(tuple+offset));
 572                count++;
 573                break;
 574            }
 575            case AGG_COUNT:
 576            {
 577                count++;
 578                break;
 579            }
 580          }
 581        }
 582        if( AGG_AVG == aType) AllDataType::divVal(buf, &count, type);
 583        else if (AGG_COUNT == aType) (*(int*)buf) = count;
 584        delete info;
 585        return OK;
 586    }
 587
 588    char *tuple = (char*) fetchNoBind(rv);
 589    if ( NULL == tuple)  return OK;
 590    int count =1;
 591    AllDataType::copyVal(buf, (void*) (tuple+offset), type, length);
 592    while(1) {
 593        tuple = (char*) fetchNoBind(rv);
 594        if (NULL == tuple) break;
 595        switch(aType) {
 596            case AGG_MIN:
 597            {
 598                res = AllDataType::compareVal(buf, (void*) (tuple+offset),
 599                                OpGreaterThanEquals,
 600                                type, length);
 601                if (res) AllDataType::copyVal(buf, (void*) (tuple+offset),
 602                                      type, length);
 603                break;
 604            }
 605            case AGG_MAX:
 606            {
 607                res = AllDataType::compareVal(buf, (void*) (tuple+offset),
 608                                OpLessThanEquals,
 609                                type, length);
 610                if (res) AllDataType::copyVal(buf, (void*) (tuple+offset),
 611                                      type, length);
 612                break;
 613            }
 614            case AGG_SUM:
 615            {
 616                AllDataType::addVal(buf, (void*) (tuple+offset),
 617                                type);
 618                break;
 619            }
 620            case AGG_AVG:
 621            {
 622                AllDataType::addVal(buf, (void*) (tuple+offset),
 623                                type);
 624                count++;
 625                break;
 626            }
 627            case AGG_COUNT:
 628            {
 629                count++;
 630                break;
 631            }
 632        }
 633    }
 634    switch(aType) {
 635        case AGG_AVG:
 636        {
 637            AllDataType::divVal(buf, &count,type);
 638            break;
 639        }
 640        case AGG_COUNT:
 641        {
 642            (*(int*)buf) = count;
 643            break;
 644        }
 645    }
 646    delete info;
 647    return OK;
 648 }
 649 DbRetVal TableImpl::insertTuple()
 650 {
 651     DbRetVal ret =OK;
 652     void *tptr = ((Chunk*)chunkPtr_)->allocate(db_, &ret);
 653     if (NULL == tptr)
 654     {
 655         printError(ret, "Unable to allocate record from chunk");
 656         return ret;
 657     }
 658     ret = lMgr_->getExclusiveLock(tptr, trans);
 659     if (OK != ret)
 660     {
 661         ((Chunk*)chunkPtr_)->free(db_, tptr);
 662         printError(ret, "Could not get lock for the insert tuple %x", tptr);
 663         return ErrLockTimeOut;
 664     }
 665
 666     curTuple_ = tptr;
 667
 668     ret = copyValuesFromBindBuffer(tptr);
 669     if (ret != OK)
 670     {
 671         printError(ret, "Unable to copy values from bind buffer");
 672         (*trans)->removeFromHasList(db_, tptr);
 673         lMgr_->releaseLock(tptr);
 674         ((Chunk*)chunkPtr_)->free(db_, tptr);
 675         return ret;
 676     }
 677     int addSize = 0;
 678     if (numFlds_ < 31)
 679     {
 680         addSize = 4;
 681         *(int*)((char*)(tptr) + (length_-addSize)) = iNullInfo;
 682     }
 683     else
 684     {
 685         addSize = os::align(numFlds_);
 686         os::memcpy(((char*)(tptr) + (length_-addSize)), cNullInfo, addSize);
 687
 688     }
 689     //int tupleSize = length_ + addSize;
 690     if (NULL != indexPtr_)
 691     {
 692         int i;
 693         //it has index
 694         for (i = 0; i < numIndexes_ ; i++)
 695         {
 696             ret = insertIndexNode(*trans, indexPtr_[i], idxInfo[i], tptr);
 697             if (ret != OK) { printError(ret, "Error in inserting to index"); break;}
 698         }
 699         if (i != numIndexes_ )
 700         {
 701             for (int j = 0; j < i ; j++) {
 702                 printError(ErrWarning, "Deleting index node");
 703                 deleteIndexNode(*trans, indexPtr_[j], idxInfo[j], tptr);
 704             }
 705             lMgr_->releaseLock(tptr);
 706             (*trans)->removeFromHasList(db_, tptr);
 707             ((Chunk*)chunkPtr_)->free(db_, tptr);
 708             printError(ret, "Unable to insert index node for tuple %x ", tptr);
 709             return ret;
 710         }
 711     }
 712     if (undoFlag)
 713         ret = (*trans)->appendUndoLog(sysDB_, InsertOperation, tptr, length_);
 714     if (ret != OK) {
 715         printError(ret, "Unable to create undo log for %x %d", tptr, *(int*)tptr);
 716         for (int j = 0; j < numIndexes_ ; j++) {
 717             printError(ErrWarning, "Deleting index node");
 718             deleteIndexNode(*trans, indexPtr_[j], idxInfo[j], tptr);
 719         }
 720         lMgr_->releaseLock(tptr);
 721         (*trans)->removeFromHasList(db_, tptr);
 722         ((Chunk*)chunkPtr_)->free(db_, tptr);
 723     }
 724     return ret;
 725 }
 726
 727 DbRetVal TableImpl::deleteTuple()
 728 {
 729     if (NULL == curTuple_)
 730     {
 731         printError(ErrNotOpen, "Scan not open: No Current tuple");
 732         return ErrNotOpen;
 733     }
 734     DbRetVal ret = lMgr_->getExclusiveLock(curTuple_, trans);
 735     if (OK != ret)
 736     {
 737         printError(ret, "Could not get lock for the delete tuple %x", curTuple_);
 738         return ErrLockTimeOut;
 739     }
 740
 741     if (NULL != indexPtr_)
 742     {
 743         int i;
 744         //it has index
 745         for (i = 0; i < numIndexes_ ; i++)
 746         {
 747             ret = deleteIndexNode(*trans, indexPtr_[i], idxInfo[i], curTuple_);
 748             if (ret != OK) break;
 749         }
 750         if (i != numIndexes_ )
 751         {
 752             for (int j = 0; j < i ; j++)
 753                 insertIndexNode(*trans, indexPtr_[j], idxInfo[j], curTuple_);
 754             lMgr_->releaseLock(curTuple_);
 755             (*trans)->removeFromHasList(db_, curTuple_);
 756             printError(ret, "Unable to insert index node for tuple %x", curTuple_);
 757             return ret;
 758         }
 759     }
 760     ((Chunk*)chunkPtr_)->free(db_, curTuple_);
 761     if (undoFlag)
 762         ret = (*trans)->appendUndoLog(sysDB_, DeleteOperation, curTuple_, length_);
 763     iter->prev();
 764     return ret;
 765 }
 766
 767 int TableImpl::deleteWhere()
 768 {
 769     int tuplesDeleted = 0;
 770     DbRetVal rv  = OK;
 771     rv =  execute();
 772     if (rv !=OK) return (int) rv;
 773     while(true){
 774         fetchNoBind( rv);
 775         if (rv != OK) { tuplesDeleted = (int)rv; break; }
 776         if (NULL == curTuple_) break;
 777         rv = deleteTuple();
 778         if (rv != OK) {
 779             printError(rv, "Error: Could only delete %d tuples", tuplesDeleted);
 780             closeScan();
 781             return (int) rv;
 782         }
 783         tuplesDeleted++;
 784     }
 785     closeScan();
 786     return tuplesDeleted;
 787 }
 788
 789 int TableImpl::truncate()
 790 {
 791     //take exclusive lock on the table
 792     //get the chunk ptr of the table
 793     //traverse the tablechunks and free all the pages except the first one
 794     //get the chunk ptr of all its indexes
 795     //traverse the indexchunks and free all the pages except the first one
 796     //release table lock
 797
 798     //TEMPORARY FIX
 799     DbRetVal rv = OK;
 800     Predicate* tmpPred = pred_;
 801     pred_ = NULL;
 802     isPlanCreated = false;
 803     int tuplesDeleted = deleteWhere();
 804     isPlanCreated = false;
 805     pred_ = tmpPred;
 806     return tuplesDeleted;
 807 }
 808
 809 DbRetVal TableImpl::updateTuple()
 810 {
 811     if (NULL == curTuple_)
 812     {
 813         printError(ErrNotOpen, "Scan not open: No Current tuple");
 814         return ErrNotOpen;
 815     }
 816     DbRetVal ret = lMgr_->getExclusiveLock(curTuple_, trans);
 817     if (OK != ret)
 818     {
 819         printError(ret, "Could not get lock for the update tuple %x", curTuple_);
 820         return ErrLockTimeOut;
 821     }
 822     if (NULL != indexPtr_)
 823     {
 824         //it has index
 825         //TODO::If it fails while updating index node, we have to undo all the updates
 826         //on other indexes on the table.Currently it will leave the database in an
 827         //inconsistent state.
 828         for (int i = 0; i < numIndexes_ ; i++)
 829         {
 830             ret = updateIndexNode(*trans, indexPtr_[i], idxInfo[i], curTuple_);
 831             if (ret != OK)
 832             {
 833                 lMgr_->releaseLock(curTuple_);
 834                 (*trans)->removeFromHasList(db_, curTuple_);
 835                 printError(ret, "Unable to update index node for tuple %x", curTuple_);
 836                 return ret;
 837             }
 838         }
 839     }
 840     if (undoFlag)
 841         ret = (*trans)->appendUndoLog(sysDB_, UpdateOperation, curTuple_, length_);
 842     if (ret != OK) return ret;
 843     int addSize = 0;
 844     int iNullVal=iNullInfo;
 845     if (numFlds_ < 31){
 846         addSize=4;
 847         if(!iNullVal){
 848             iNullInfo = *(int*)((char*)(curTuple_) + (length_- addSize));
 849         }
 850         else
 851         {
 852             *(int*)((char*)(curTuple_) + (length_-addSize)) |= iNullInfo;
 853         }
 854     }
 855     DbRetVal rv = copyValuesFromBindBuffer(curTuple_, false);
 856     if (rv != OK) {
 857         lMgr_->releaseLock(curTuple_);
 858         (*trans)->removeFromHasList(db_, curTuple_);
 859         return rv;
 860     }
 861
 862     if (numFlds_ < 31)
 863     {
 864         if (!iNullVal) {
 865                 *(int*)((char*)(curTuple_) + (length_-addSize)) = iNullInfo;
 866                 iNullInfo=0;
 867         }
 868         else iNullInfo=iNullVal;
 869     }
 870     else
 871     {
 872         addSize = os::align(numFlds_);
 873         //TODO::Do not do blind memcpy. It should OR each and every char
 874         //os::memcpy(((char*)(curTuple_) + (length_-addSize)), cNullInfo, addSize);
 875
 876     }
 877     return OK;
 878 }
 879
 880 void TableImpl::printInfo()
 881 {
 882     printf("  <TableName> %s </TableName>\n", tblName_);
 883     printf("  <TupleCount> %d </TupleCount>\n", numTuples());
 884     printf("  <PagesUsed> %d </PagesUsed>\n", pagesUsed());
 885     printf("  <SpaceUsed> %d </SpaceUsed>\n", spaceUsed());
 886     printf("  <Indexes> %d <Indexes>\n", numIndexes_);
 887     printf("  <TupleLength> %d </TupleLength>\n", length_);
 888     printf("  <Fields> %d </Fields>\n", numFlds_);
 889     printf("  <Indexes>\n");
 890     for (int i =0; i<numIndexes_; i++)
 891         printf("<IndexName> %s </IndexName>\n", CatalogTableINDEX::getName(indexPtr_[i]));
 892     printf("  </Indexes>\n");
 893
 894 }
 895
 896 DbRetVal TableImpl::copyValuesFromBindBuffer(void *tuplePtr, bool isInsert)
 897 {
 898     //Iterate through the bind list and copy the value here
 899     FieldIterator fIter = fldList_.getIterator();
 900     char *colPtr = (char*) tuplePtr;
 901     int fldpos=1;
 902     while (fIter.hasElement())
 903     {
 904         FieldDef *def = fIter.nextElement();
 905         if(def->isAutoIncrement_ && isInsert)
 906         {
 907             void *dest = AllDataType::alloc(def->type_, def->length_);
 908             AllDataType::copyVal(dest,ptrToAuto, def->type_, def->length_);
 909             if(def->bindVal_==NULL)
 910             {
 911                 AllDataType::increment(colPtr, dest , def->type_);
 912                 AllDataType::copyVal(ptrToAuto,colPtr, def->type_, def->length_);
 913                 colPtr = colPtr + def->length_;
 914                 fldpos++;
 915                 free(dest);
 916                 continue;
 917             }else {
 918                 if(AllDataType::compareVal(def->bindVal_, dest, OpGreaterThan, def->type_)){
 919                     AllDataType::copyVal(ptrToAuto,def->bindVal_, def->type_, def->length_);
 920                 }
 921                 free(dest);
 922             }
 923         }
 924         if (def->isNull_ && !def->isDefault_ && NULL == def->bindVal_ && isInsert)
 925         {
 926             printError(ErrNullViolation, "NOT NULL constraint violation for field %s\n", def->fldName_);
 927             return ErrNullViolation;
 928         }
 929         if (def->isDefault_ && NULL == def->bindVal_ && isInsert)
 930         {
 931             void *dest = AllDataType::alloc(def->type_, def->length_);
 932             AllDataType::convert(typeString, def->defaultValueBuf_, def->type_, dest, def->length_);
 933             AllDataType::copyVal(colPtr, dest, def->type_, def->length_);
 934             colPtr = colPtr + def->length_;
 935             fldpos++;
 936             free (dest);
 937             continue;
 938         }
 939         switch(def->type_)
 940         {
 941             case typeString:
 942                 if (NULL != def->bindVal_)
 943                 {
 944                     if(!isInsert && isFldNull(fldpos)){clearNullBit(fldpos);}
 945                     strcpy((char*)colPtr, (char*)def->bindVal_);
 946                     *(((char*)colPtr) + (def->length_-1)) = '\0';
 947                 }
 948                 else if (!def->isNull_ && !def->bindVal_ && isInsert)  setNullBit(fldpos);
 949                 colPtr = colPtr + def->length_;
 950                 break;
 951             case typeBinary:
 952                 if (NULL != def->bindVal_ )
 953                 {
 954                     if(!isInsert && isFldNull(fldpos)){clearNullBit(fldpos);}
 955                     DbRetVal rv = AllDataType::strToValue(colPtr, (char *) def->bindVal_, def->type_, def->length_);
 956                     if (rv != OK) return ErrBadArg;
 957                 }
 958                 else if (!def->isNull_ && isInsert && !def->bindVal_)  setNullBit(fldpos);
 959                 colPtr = colPtr + def->length_;
 960                 break;
 961             default:
 962                 if (NULL != def->bindVal_){
 963                     if(!isInsert && isFldNull(fldpos)){clearNullBit(fldpos);}
 964                     AllDataType::copyVal(colPtr, def->bindVal_, def->type_);}
 965                 else { if (!def->isNull_ && isInsert)  setNullBit(fldpos); }
 966                 colPtr = colPtr + def->length_;
 967                 break;
 968         }
 969         fldpos++;
 970     }
 971     return OK;
 972 }
 973 void TableImpl::clearNullBit(int fldpos)
 974 {
 975     if (isIntUsedForNULL){
 976         CLEARBIT(iNullInfo, fldpos);}
 977     else
 978         cNullInfo[fldpos-1] = 0;
 979 }
 980 void TableImpl::setNullBit(int fldpos)
 981 {
 982     if (isIntUsedForNULL)
 983         SETBIT(iNullInfo, fldpos);
 984     else
 985         cNullInfo[fldpos-1] = 1;
 986 }
 987 DbRetVal TableImpl::copyValuesToBindBuffer(void *tuplePtr)
 988 {
 989     //Iterate through the bind list and copy the value here
 990     char *colPtr = (char*) tuplePtr;
 991     FieldDef *def = NULL;
 992     for (int i = 0; i < numBindFlds_; i++) {
 993        def = (FieldDef *) bindListArray_[i];
 994        colPtr =  (char *) tuplePtr + def->offset_;
 995        AllDataType::copyVal(def->bindVal_, colPtr, def->type_, def->length_);
 996     }
 997     return OK;
 998 }
 999
1000 //-1 index not supported
1001 DbRetVal TableImpl::insertIndexNode(Transaction *tr, void *indexPtr, IndexInfo *info, void *tuple)
1002 {
1003     CINDEX *iptr = (CINDEX*)indexPtr;
1004     DbRetVal ret = OK;
1005     printDebug(DM_Table, "Inside insertIndexNode type %d", iptr->indexType_);
1006     Index* idx = Index::getIndex(iptr->indexType_);
1007     ret = idx->insert(this, tr, indexPtr, info, tuple,undoFlag);
1008     return ret;
1009 }
1010
1011 DbRetVal TableImpl::deleteIndexNode(Transaction *tr, void *indexPtr, IndexInfo *info, void *tuple)
1012 {
1013     CINDEX *iptr = (CINDEX*)indexPtr;
1014     DbRetVal ret = OK;
1015     Index* idx = Index::getIndex(iptr->indexType_);
1016     ret = idx->remove(this, tr, indexPtr, info, tuple, undoFlag);
1017     return ret;
1018 }
1019 void TableImpl::printSQLIndexString()
1020 {
1021     CatalogTableINDEXFIELD cIndexField(sysDB_);
1022     char fName[IDENTIFIER_LENGTH];
1023     char *fldName = fName;
1024     char idxName[IDENTIFIER_LENGTH];
1025     DataType type;
1026     for (int i = 0; i < numIndexes_ ; i++)
1027     {
1028         CINDEX *iptr = (CINDEX*) indexPtr_[i];
1029         sprintf(idxName,"%s_idx_Auto_increment",getName());
1030         if(strcmp(iptr->indName_,idxName)==0){ continue; }
1031         printf("CREATE INDEX %s on %s ( ", iptr->indName_, getName());
1032         FieldList fldList;
1033         cIndexField.getFieldInfo(iptr, fldList);
1034         FieldIterator fIter = fldList.getIterator();
1035         bool firstFld = true;
1036         while(fIter.hasElement())
1037         {
1038             FieldDef *def = fIter.nextElement();
1039             if (firstFld) { printf(" %s ", def->fldName_); firstFld = false; }
1040             else printf(" ,%s ", def->fldName_);
1041         }
1042         printf(" ) ");
1043         if (iptr->indexType_ == hashIndex) printf(" HASH ");
1044         else printf(" TREE ");
1045         if (((HashIndexInfo*) idxInfo[i])->isUnique) printf(" UNIQUE");
1046         if(((HashIndexInfo*) idxInfo[i])->noOfBuckets != 1009 ) printf(" SIZE %d ",((HashIndexInfo*) idxInfo[i])->noOfBuckets );
1047         printf(";\n");
1048     }
1049 }
1050
1051
1052 DbRetVal TableImpl::updateIndexNode(Transaction *tr, void *indexPtr, IndexInfo *info, void *tuple)
1053 {
1054     CINDEX *iptr = (CINDEX*)indexPtr;
1055     DbRetVal ret = OK;
1056     Index* idx = Index::getIndex(iptr->indexType_);
1057     //TODO::currently it updates irrespective of whether the key changed or not
1058     //because of this commenting the whole index update code. relook at it and uncomment
1059
1060     ret = idx->update(this, tr, indexPtr, info, tuple, undoFlag);
1061
1062     return ret;
1063 }
1064
1065
1066 void TableImpl::setTableInfo(char *name, int tblid, size_t  length,
1067                        int numFld, int numIdx, void *chunk)
1068 {
1069     strcpy(tblName_, name);
1070     tblID_ = tblid;
1071     length_ = length;
1072     numFlds_ = numFld;
1073     numIndexes_ = numIdx;
1074     chunkPtr_ = chunk;
1075 }
1076
1077 long TableImpl::spaceUsed()
1078 {
1079     Chunk *chk = (Chunk*)chunkPtr_;
1080     long totSize = chk->getTotalDataNodes() * chk->getSize();
1081     totSize = totSize + (chk->totalPages() * sizeof (PageInfo));
1082     return totSize;
1083 }
1084
1085 int TableImpl::pagesUsed()
1086 {
1087     Chunk *chk = (Chunk*)chunkPtr_;
1088     return chk->totalPages();
1089 }
1090
1091 long TableImpl::numTuples()
1092 {
1093     return ((Chunk*)chunkPtr_)->getTotalDataNodes();
1094 }
1095
1096 List TableImpl::getFieldNameList()
1097 {
1098     List fldNameList;
1099     FieldIterator fIter = fldList_.getIterator();
1100     char fieldName[IDENTIFIER_LENGTH];
1101     while (fIter.hasElement())
1102     {
1103         FieldDef *def = fIter.nextElement();
1104         Identifier *elem = new Identifier();
1105         Table::getFieldNameAlone(def->fldName_, fieldName);
1106         sprintf(elem->name, "%s.%s", getName(), fieldName);
1107         fldNameList.append(elem);
1108     }
1109     return fldNameList;
1110 }
1111 DbRetVal TableImpl::close()
1112 {
1113     if (iter) { iter->close(); delete iter; iter = NULL; }
1114     printDebug(DM_Database,"Closing table handle: %x", this);
1115     //table->unlock();
1116     delete pred_;
1117     delete this;
1118     logFinest(logger, "Closing Table");
1119     return OK;
1120 }
1121
1122 DbRetVal TableImpl::closeScan()
1123 {
1124     //do not throw scan not open error
1125     //this function will be called by table handle
1126     if (iter) {
1127         // iter->reset();
1128         //PRABA::TEMP::otherwise fails.check with kishor
1129         delete iter;
1130         iter = NULL;
1131     }
1132     return OK;
1133 }
1134 DbRetVal TableImpl::lock(bool shared)
1135 {
1136
1137     DbRetVal ret = OK;
1138 /*
1139     if (shared)
1140         ret = lMgr_->getSharedLock(chunkPtr_, NULL);
1141     else
1142         ret = lMgr_->getExclusiveLock(chunkPtr_, NULL);
1143     if (OK != ret)
1144     {
1145         printError(ret, "Could not exclusive lock on the table %x", chunkPtr_);
1146     }else {
1147         //do not append for S to X upgrade
1148         if (!ProcessManager::hasLockList.exists(chunkPtr_))
1149             ProcessManager::hasLockList.append(chunkPtr_);
1150     }
1151 */
1152     return ret;
1153 }
1154 DbRetVal TableImpl::unlock()
1155 {
1156 /*
1157     if (!ProcessManager::hasLockList.exists(chunkPtr_)) return OK;
1158     DbRetVal ret = lMgr_->releaseLock(chunkPtr_);
1159     if (OK != ret)
1160     {
1161         printError(ret, "Could not release exclusive lock on the table %x", chunkPtr_);
1162     }else
1163     {
1164         ProcessManager::hasLockList.remove(chunkPtr_);
1165     }
1166 */
1167     return OK;
1168 }
1169
1170 TableImpl::~TableImpl()
1171 {
1172     if (NULL != iter ) { delete iter; iter = NULL; }
1173     if (NULL != indexPtr_) { delete[] indexPtr_; indexPtr_ = NULL;  }
1174     if (NULL != idxInfo)
1175     {
1176         for (int i = 0; i < numIndexes_; i++) delete idxInfo[i];
1177         delete[] idxInfo;
1178         idxInfo = NULL;
1179     }
1180     if (numFlds_ > 31 && cNullInfo != NULL) { free(cNullInfo); cNullInfo = NULL; }
1181     if (bindList_.size()) bindList_.reset();
1182     if (bindListArray_) { free (bindListArray_); bindListArray_ = NULL; }
1183     fldList_.removeAll();
1184
1185 }
1186
1187 void *TableImpl::getBindFldAddr(const char *name)
1188 {
1189         return fldList_.getBindField(name);
1190 }
1191 bool TableImpl::isTableInvolved(char  *tblName)
1192 {
1193     //printf("Table isTableInvolved called for %s with %s\n", tblName, getName());
1194     if (0 == strcmp(getName(), tblName)) return true; else return false;
1195 }
1196 bool TableImpl::pushPredicate(Predicate *pred)
1197 {
1198     bool ret = false;
1199     PredicateImpl *pImpl = (PredicateImpl*) pred;
1200     char tableName[IDENTIFIER_LENGTH];
1201     Table::getTableNameAlone(pImpl->getFldName1(), tableName);
1202     //printf("predicate tbl name %s\n", tableName);
1203
1204     //if predicate is of form t1.f1=t2.f1 then do not push here
1205     if (0 != strcmp(pImpl->getFldName2(),"")) return ret;
1206
1207     if (0 == strcmp(getName(), tableName))
1208     {
1209         setPredicate(pred);
1210         //printf("PRABA::pushed predicate in tablehdl  %s\n", getName());
1211         ret = true;
1212     }
1213     return ret;
1214 }
1215 void TableImpl::setPredicate(Predicate *pred)
1216 {
1217     if (NULL == pred_) { pred_ = pred; return; }
1218     Predicate *curPred = pred_;
1219     PredicateImpl *newPred = new PredicateImpl();
1220     newPred->setTerm(curPred, OpAnd, pred);
1221     newPred->setTable(this);
1222     pred_ = newPred;
1223     return;
1224 }
1225 void TableImpl::printPlan(int space)
1226 {
1227     char spaceBuf[IDENTIFIER_LENGTH];
1228     memset(spaceBuf, 32, IDENTIFIER_LENGTH);
1229     spaceBuf[space] = '\0';
1230     printf("%s <TABLE-NODE>\n", spaceBuf);
1231     printf("%s <NAME> %s </NAME>\n", spaceBuf, getName());
1232     printf("%s <ScanType> %s </ScanType>\n", spaceBuf, ScanTypeNames[scanType_]);
1233     PredicateImpl *pred = (PredicateImpl*)pred_;
1234     if (pred) pred->print(space+2);
1235     printf("%s </TABLE-NODE>\n", spaceBuf);
1236 }
1237