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 void TableImpl::addPredicate(char *fName, ComparisionOp op, void *buf)
 186 {
 187     //char fieldName[IDENTIFIER_LENGTH];
 188     //Table::getFieldNameAlone(fName, fieldName);
 189     PredicateImpl *pred = (PredicateImpl*) pred_;
 190     PredicateImpl *newPred = new PredicateImpl();
 191     newPred->setTerm(fName, op, buf);
 192     if (NULL == pred) { pred_ = newPred; return; }
 193     PredicateImpl *bothPred = new PredicateImpl();
 194     bothPred->setTerm(pred, OpAnd, newPred);
 195     pred_ = bothPred;
 196 }
 197
 198 DbRetVal TableImpl::optimize()
 199 {
 200     //table ptr is set in predicate because it needs to access the
 201     //type and length to evaluate
 202     if( NULL != pred_)
 203     {
 204         PredicateImpl *pred = (PredicateImpl*) pred_;
 205         pred->setTable(this);
 206         pred->setProjectionList(NULL);
 207         pred->setOffsetAndType();
 208     }
 209     return createPlan();
 210 }
 211
 212 DbRetVal TableImpl::execute()
 213 {
 214     if (NULL != iter)
 215     {
 216          printError(ErrAlready,"Scan already open:Close and re execute");
 217          return ErrAlready;
 218     }
 219     DbRetVal ret = OK;
 220     ret = optimize();
 221     if (OK != ret)
 222     {
 223         printError(ErrSysInternal,"Unable to create the plan");
 224         return ErrSysInternal;
 225     }
 226     if (useIndex_ >= 0)
 227         iter = new TupleIterator(pred_, scanType_, idxInfo[useIndex_], chunkPtr_, sysDB_->procSlot,isBetween,isPointLook);
 228     else if (scanType_ == fullTableScan)
 229         iter = new TupleIterator(pred_, scanType_, NULL, chunkPtr_, sysDB_->procSlot,isBetween,isPointLook);
 230     else
 231     {
 232         printError(ErrSysFatal,"Unable to create tuple iterator");//should never happen
 233         return ErrSysFatal;
 234     }
 235     ret = iter->open();
 236     if (OK != ret)
 237     {
 238         printError(ret,"Unable to open the iterator");
 239         return ret;
 240     }
 241     return OK;
 242 }
 243
 244
 245 DbRetVal TableImpl::createPlan()
 246 {
 247     if (isPlanCreated) {
 248         //will do early return here. plan is generated only when setPredicate is called.
 249         if (scanType_ == unknownScan) return ErrSysFatal; //this should never happen
 250         else return OK;
 251     }
 252     isBetween=false;
 253     isPointLook = false;
 254     useIndex_ = -1;
 255
 256     FieldIterator fIter = fldList_.getIterator();
 257     FieldDef *def = NULL;
 258     while ((def = fIter.nextElement())!= NULL) {
 259         if (NULL != def->bindVal_) bindList_.append(def);
 260     }
 261     numBindFlds_ = bindList_.size();
 262     bindListArray_ = (void **) malloc(numBindFlds_ * sizeof (void *));
 263     void *elem = NULL;
 264     int i = 0;
 265     ListIterator it = bindList_.getIterator();
 266     while ((elem = it.nextElement()) != NULL) bindListArray_[i++] = elem;
 267
 268     //if there are no predicates then go for full scan
 269     //if there are no indexes then go for full scan
 270     if (NULL == pred_ || NULL == indexPtr_)
 271     {
 272         scanType_ = fullTableScan;
 273         isPlanCreated = true;
 274         return OK;
 275     }
 276     if (NULL != indexPtr_)
 277     {
 278        PredicateImpl *pred = (PredicateImpl*)pred_;
 279        printDebug(DM_Predicate, "predicate does not involve NOT , OR operator");
 280        if (!pred->isNotOrInvolved())
 281        {
 282            printDebug(DM_Predicate, "predicate does not involve NOT , OR operator");
 283           for (int i =0; i < numIndexes_; i++)
 284           {
 285               HashIndexInfo* info = (HashIndexInfo*) idxInfo[i];
 286               FieldIterator iter = info->idxFldList.getIterator();
 287               while(iter.hasElement())
 288               {
 289                 FieldDef *def = iter.nextElement();
 290                 if (pred->pointLookupInvolved(def->fldName_))
 291                 {
 292                   printDebug(DM_Predicate, "point lookup involved for field %s",def->fldName_);
 293                   if(hashIndex == info->indType) scanType_ = hashIndexScan;
 294                   else scanType_ = treeIndexScan;
 295                   isPlanCreated = true;
 296                   isPointLook = true;
 297                   useIndex_ = i;
 298                 }
 299                 else if (pred->isBetweenInvolved(def->fldName_))
 300                 {
 301                     if (treeIndex == info->indType)
 302                     {
 303                      scanType_ = treeIndexScan;
 304                      isPlanCreated = true;
 305                      useIndex_ = i;
 306                      isBetween=true;
 307                      break; //no composite index for tree index
 308                     }
 309                 }
 310                 else if (pred->rangeQueryInvolved(def->fldName_))
 311                 {
 312                   printDebug(DM_Predicate, "range lookup involved for field %s",def->fldName_);
 313                   if (treeIndex == info->indType)
 314                   {
 315                      scanType_ = treeIndexScan;
 316                      isPlanCreated = true;
 317                      useIndex_ = i;
 318                      break; //no composite index for tree index
 319                   }
 320                 }else {
 321                     useIndex_ = -1;
 322                     break;
 323                 }
 324              }//while iter.hasElement()
 325              if (useIndex_ != -1) return OK;
 326            }//for
 327         }
 328     }
 329     scanType_ = fullTableScan;
 330     isPlanCreated = true;
 331     return OK;
 332 }
 333
 334 void* TableImpl::fetch()
 335 {
 336     fetchNoBind();
 337     if (NULL == curTuple_) return curTuple_;
 338     copyValuesToBindBuffer(curTuple_);
 339     return curTuple_;
 340 }
 341 void* TableImpl::fetch(DbRetVal &rv)
 342 {
 343     fetchNoBind(rv);
 344     if (NULL == curTuple_) return curTuple_;
 345     copyValuesToBindBuffer(curTuple_);
 346     return curTuple_;
 347 }
 348
 349 void* TableImpl::fetchNoBind()
 350 {
 351     if (NULL == iter)
 352     {
 353         printError(ErrNotOpen,"Scan not open or Scan is closed\n");
 354         return NULL;
 355     }
 356     void *prevTuple = curTuple_;
 357     curTuple_ = iter->next();
 358     if (NULL == curTuple_)
 359     {
 360         return NULL;
 361     }
 362     DbRetVal lockRet = OK;
 363     if ((*trans)->isoLevel_ == READ_COMMITTED)
 364     {
 365         //if iso level is read committed, operation duration lock is sufficent
 366         //so release it here itself.
 367         int tries = 5;
 368         struct timeval timeout;
 369         timeout.tv_sec = Conf::config.getMutexSecs();
 370         timeout.tv_usec = Conf::config.getMutexUSecs();
 371
 372         bool status = false;
 373         while(true) {
 374             lockRet = lMgr_->isExclusiveLocked( curTuple_, trans, status);
 375             if (OK != lockRet)
 376             {
 377                 printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 378                 curTuple_ = prevTuple;
 379                 return NULL;
 380             }
 381             if (!status) break;
 382             tries--;
 383             if (tries == 0) break;
 384             os::select(0, 0, 0, 0, &timeout);
 385
 386         }
 387         if (tries == 0)
 388         {
 389             printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 390             curTuple_ = prevTuple;
 391             return NULL;
 392         }
 393     }
 394     else if ((*trans)->isoLevel_ == READ_REPEATABLE) {
 395         lockRet = lMgr_->getSharedLock(curTuple_, trans);
 396         if (OK != lockRet)
 397         {
 398             printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 399             curTuple_ = prevTuple;
 400             return NULL;
 401         }
 402
 403     }
 404     return curTuple_;
 405 }
 406
 407 void* TableImpl::fetchNoBind(DbRetVal &rv)
 408 {
 409     rv = OK;
 410     if (NULL == iter)
 411     {
 412         printError(ErrNotOpen,"Scan not open or Scan is closed\n");
 413         rv = ErrNotOpen;
 414         return NULL;
 415     }
 416     void *prevTuple = curTuple_;
 417     curTuple_ = iter->next();
 418     if (NULL == curTuple_)
 419     {
 420         return NULL;
 421     }
 422     DbRetVal lockRet = OK;
 423     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             rv = ErrLockTimeOut;
 429             curTuple_ = prevTuple;
 430             return NULL;
 431         }
 432
 433     }
 434     else if ((*trans)->isoLevel_ == READ_COMMITTED)
 435     {
 436         //if iso level is read committed, operation duration lock is sufficent
 437         //so release it here itself.
 438         int tries = 5;
 439         struct timeval timeout;
 440         timeout.tv_sec = Conf::config.getMutexSecs();
 441         timeout.tv_usec = Conf::config.getMutexUSecs();
 442
 443         bool status = false;
 444         while(true) {
 445             lockRet = lMgr_->isExclusiveLocked( curTuple_, trans, status);
 446             if (OK != lockRet)
 447             {
 448                 printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 449                 curTuple_ = prevTuple;
 450                 rv = ErrLockTimeOut;
 451                 return NULL;
 452             }
 453             if (!status) break;
 454             tries--;
 455             if (tries == 0) break;
 456             os::select(0, 0, 0, 0, &timeout);
 457
 458         }
 459         if (tries == 0)
 460         {
 461             printError(lockRet, "Unable to get the lock for the tuple %x", curTuple_);
 462             curTuple_ = prevTuple;
 463             rv = ErrLockTimeOut;
 464             return NULL;
 465         }
 466     }
 467     return curTuple_;
 468 }
 469 DbRetVal TableImpl::fetchAgg(const char * fldName, AggType aType, void *buf)
 470 {
 471    FieldInfo *info = new FieldInfo();
 472    DbRetVal rv = getFieldInfo(fldName, info);
 473    if (OK != rv) return rv;
 474    bool res= false;
 475
 476    char *tuple = (char*) fetchNoBind(rv);
 477    if ( NULL == tuple)
 478    {
 479       *(int*)buf = 0; //assuming int. could create porting problems(64 |endian)
 480       return OK;
 481    }
 482    int count =1;
 483    AllDataType::copyVal(buf, (void*) (tuple+info->offset), info->type, info->length);
 484    while(1) {
 485        tuple = (char*) fetchNoBind(rv);
 486        if (NULL == tuple) break;
 487        switch(aType) {
 488            case AGG_MIN:
 489            {
 490                res = AllDataType::compareVal(buf, (void*) (tuple+info->offset),
 491                                OpGreaterThan,
 492                                info->type, info->length);
 493                if (res) AllDataType::copyVal(buf, (void*) (tuple+info->offset),
 494                                      info->type, info->length);
 495                break;
 496            }
 497            case AGG_MAX:
 498            {
 499                res = AllDataType::compareVal(buf, (void*) (tuple+info->offset),
 500                                OpLessThan,
 501                                info->type, info->length);
 502                if (res) AllDataType::copyVal(buf, (void*) (tuple+info->offset),
 503                                      info->type, info->length);
 504                break;
 505            }
 506            case AGG_SUM:
 507            {
 508                AllDataType::addVal(buf, (void*) (tuple+info->offset),
 509                                info->type);
 510                break;
 511            }
 512            case AGG_AVG:
 513            {
 514                AllDataType::addVal(buf, (void*) (tuple+info->offset),
 515                                info->type);
 516                count++;
 517                break;
 518            }
 519            case AGG_COUNT:
 520            {
 521                count++;
 522                break;
 523            }
 524        }
 525    }
 526    switch(aType) {
 527        case AGG_AVG:
 528        {
 529            AllDataType::divVal(buf, &count,info->type);
 530            break;
 531        }
 532        case AGG_COUNT:
 533        {
 534            (*(int*)buf) = count;
 535            break;
 536        }
 537    }
 538    delete info;
 539    return OK;
 540 }
 541 DbRetVal TableImpl::insertTuple()
 542 {
 543     DbRetVal ret =OK;
 544     void *tptr = ((Chunk*)chunkPtr_)->allocate(db_, &ret);
 545     if (NULL == tptr)
 546     {
 547         printError(ret, "Unable to allocate record from chunk");
 548         return ret;
 549     }
 550     ret = lMgr_->getExclusiveLock(tptr, trans);
 551     if (OK != ret)
 552     {
 553         ((Chunk*)chunkPtr_)->free(db_, tptr);
 554         printError(ret, "Could not get lock for the insert tuple %x", tptr);
 555         return ErrLockTimeOut;
 556     }
 557
 558     curTuple_ = tptr;
 559
 560     ret = copyValuesFromBindBuffer(tptr);
 561     if (ret != OK)
 562     {
 563         printError(ret, "Unable to copy values from bind buffer");
 564         (*trans)->removeFromHasList(db_, tptr);
 565         lMgr_->releaseLock(tptr);
 566         ((Chunk*)chunkPtr_)->free(db_, tptr);
 567         return ret;
 568     }
 569     int addSize = 0;
 570     if (numFlds_ < 31)
 571     {
 572         addSize = 4;
 573         *(int*)((char*)(tptr) + (length_-addSize)) = iNullInfo;
 574     }
 575     else
 576     {
 577         addSize = os::align(numFlds_);
 578         os::memcpy(((char*)(tptr) + (length_-addSize)), cNullInfo, addSize);
 579
 580     }
 581     //int tupleSize = length_ + addSize;
 582     if (NULL != indexPtr_)
 583     {
 584         int i;
 585         //it has index
 586         for (i = 0; i < numIndexes_ ; i++)
 587         {
 588             ret = insertIndexNode(*trans, indexPtr_[i], idxInfo[i], tptr);
 589             if (ret != OK) { printError(ret, "Error in inserting to index"); break;}
 590         }
 591         if (i != numIndexes_ )
 592         {
 593             for (int j = 0; j < i ; j++) {
 594                 printError(ErrWarning, "Deleting index node");
 595                 deleteIndexNode(*trans, indexPtr_[j], idxInfo[j], tptr);
 596             }
 597             lMgr_->releaseLock(tptr);
 598             (*trans)->removeFromHasList(db_, tptr);
 599             ((Chunk*)chunkPtr_)->free(db_, tptr);
 600             //PRABA::TEMP
 601             //printError(ret, "Unable to insert index node for tuple %x ", tptr);
 602             printError(ret, "Unable to insert index node for tuple %x %d", tptr, *(int*)tptr);
 603             return ret;
 604         }
 605     }
 606     if (undoFlag)
 607         ret = (*trans)->appendUndoLog(sysDB_, InsertOperation, tptr, length_);
 608     if (ret != OK) {
 609         printError(ret, "Unable to create undo log for %x %d", tptr, *(int*)tptr);
 610         for (int j = 0; j < numIndexes_ ; j++) {
 611             printError(ErrWarning, "Deleting index node");
 612             deleteIndexNode(*trans, indexPtr_[j], idxInfo[j], tptr);
 613         }
 614         lMgr_->releaseLock(tptr);
 615         (*trans)->removeFromHasList(db_, tptr);
 616         ((Chunk*)chunkPtr_)->free(db_, tptr);
 617     }
 618     return ret;
 619 }
 620
 621 DbRetVal TableImpl::deleteTuple()
 622 {
 623     if (NULL == curTuple_)
 624     {
 625         printError(ErrNotOpen, "Scan not open: No Current tuple");
 626         return ErrNotOpen;
 627     }
 628     DbRetVal ret = lMgr_->getExclusiveLock(curTuple_, trans);
 629     if (OK != ret)
 630     {
 631         printError(ret, "Could not get lock for the delete tuple %x", curTuple_);
 632         return ErrLockTimeOut;
 633     }
 634
 635     if (NULL != indexPtr_)
 636     {
 637         int i;
 638         //it has index
 639         for (i = 0; i < numIndexes_ ; i++)
 640         {
 641             ret = deleteIndexNode(*trans, indexPtr_[i], idxInfo[i], curTuple_);
 642             if (ret != OK) break;
 643         }
 644         if (i != numIndexes_ )
 645         {
 646             for (int j = 0; j < i ; j++)
 647                 insertIndexNode(*trans, indexPtr_[j], idxInfo[j], curTuple_);
 648             lMgr_->releaseLock(curTuple_);
 649             (*trans)->removeFromHasList(db_, curTuple_);
 650             printError(ret, "Unable to insert index node for tuple %x", curTuple_);
 651             return ret;
 652         }
 653     }
 654     ((Chunk*)chunkPtr_)->free(db_, curTuple_);
 655     if (undoFlag)
 656         ret = (*trans)->appendUndoLog(sysDB_, DeleteOperation, curTuple_, length_);
 657     iter->prev();
 658     return ret;
 659 }
 660
 661 int TableImpl::deleteWhere()
 662 {
 663     int tuplesDeleted = 0;
 664     DbRetVal rv  = OK;
 665     rv =  execute();
 666     if (rv !=OK) return (int) rv;
 667     while(true){
 668         fetchNoBind( rv);
 669         if (rv != OK) { tuplesDeleted = (int)rv; break; }
 670         if (NULL == curTuple_) break;
 671         rv = deleteTuple();
 672         if (rv != OK) {
 673             printError(rv, "Error: Could only delete %d tuples", tuplesDeleted);
 674             close();
 675             return (int) rv;
 676         }
 677         tuplesDeleted++;
 678     }
 679     close();
 680     return tuplesDeleted;
 681 }
 682
 683 int TableImpl::truncate()
 684 {
 685     //take exclusive lock on the table
 686     //get the chunk ptr of the table
 687     //traverse the tablechunks and free all the pages except the first one
 688     //get the chunk ptr of all its indexes
 689     //traverse the indexchunks and free all the pages except the first one
 690     //release table lock
 691
 692     //TEMPORARY FIX
 693     DbRetVal rv = OK;
 694     Predicate* tmpPred = pred_;
 695     pred_ = NULL;
 696     isPlanCreated = false;
 697     int tuplesDeleted = deleteWhere();
 698     isPlanCreated = false;
 699     pred_ = tmpPred;
 700     return tuplesDeleted;
 701 }
 702
 703 DbRetVal TableImpl::updateTuple()
 704 {
 705     if (NULL == curTuple_)
 706     {
 707         printError(ErrNotOpen, "Scan not open: No Current tuple");
 708         return ErrNotOpen;
 709     }
 710     DbRetVal ret = lMgr_->getExclusiveLock(curTuple_, trans);
 711     if (OK != ret)
 712     {
 713         printError(ret, "Could not get lock for the update tuple %x", curTuple_);
 714         return ErrLockTimeOut;
 715     }
 716     if (NULL != indexPtr_)
 717     {
 718         //it has index
 719         //TODO::If it fails while updating index node, we have to undo all the updates
 720         //on other indexes on the table.Currently it will leave the database in an
 721         //inconsistent state.
 722         for (int i = 0; i < numIndexes_ ; i++)
 723         {
 724             ret = updateIndexNode(*trans, indexPtr_[i], idxInfo[i], curTuple_);
 725             if (ret != OK)
 726             {
 727                 lMgr_->releaseLock(curTuple_);
 728                 (*trans)->removeFromHasList(db_, curTuple_);
 729                 printError(ret, "Unable to update index node for tuple %x", curTuple_);
 730                 return ret;
 731             }
 732         }
 733     }
 734     if (undoFlag)
 735         ret = (*trans)->appendUndoLog(sysDB_, UpdateOperation, curTuple_, length_);
 736     if (ret != OK) return ret;
 737     int addSize = 0;
 738     int iNullVal=iNullInfo;
 739     if (numFlds_ < 31){
 740         addSize=4;
 741         if(!iNullVal){
 742             iNullInfo = *(int*)((char*)(curTuple_) + (length_- addSize));
 743         }
 744         else
 745         {
 746             *(int*)((char*)(curTuple_) + (length_-addSize)) |= iNullInfo;
 747         }
 748     }
 749     DbRetVal rv = copyValuesFromBindBuffer(curTuple_, false);
 750     if (rv != OK) {
 751         lMgr_->releaseLock(curTuple_);
 752         (*trans)->removeFromHasList(db_, curTuple_);
 753         return rv;
 754     }
 755
 756     if (numFlds_ < 31)
 757     {
 758         if (!iNullVal) {
 759                 *(int*)((char*)(curTuple_) + (length_-addSize)) = iNullInfo;
 760                 iNullInfo=0;
 761         }
 762         else iNullInfo=iNullVal;
 763     }
 764     else
 765     {
 766         addSize = os::align(numFlds_);
 767         //TODO::Do not do blind memcpy. It should OR each and every char
 768         //os::memcpy(((char*)(curTuple_) + (length_-addSize)), cNullInfo, addSize);
 769
 770     }
 771     return OK;
 772 }
 773
 774 void TableImpl::printInfo()
 775 {
 776     printf("  <TableName> %s </TableName>\n", tblName_);
 777     printf("  <TupleCount> %d </TupleCount>\n", numTuples());
 778     printf("  <PagesUsed> %d </PagesUsed>\n", pagesUsed());
 779     printf("  <SpaceUsed> %d </SpaceUsed>\n", spaceUsed());
 780     printf("  <Indexes> %d <Indexes>\n", numIndexes_);
 781     printf("  <TupleLength> %d </TupleLength>\n", length_);
 782     printf("  <Fields> %d </Fields>\n", numFlds_);
 783     printf("  <Indexes>\n");
 784     for (int i =0; i<numIndexes_; i++)
 785         printf("<IndexName> %s </IndexName>\n", CatalogTableINDEX::getName(indexPtr_[i]));
 786     printf("  </Indexes>\n");
 787
 788 }
 789
 790 DbRetVal TableImpl::copyValuesFromBindBuffer(void *tuplePtr, bool isInsert)
 791 {
 792     //Iterate through the bind list and copy the value here
 793     FieldIterator fIter = fldList_.getIterator();
 794     char *colPtr = (char*) tuplePtr;
 795     int fldpos=1;
 796     while (fIter.hasElement())
 797     {
 798         FieldDef *def = fIter.nextElement();
 799         if (def->isNull_ && !def->isDefault_ && NULL == def->bindVal_ && isInsert)
 800         {
 801             printError(ErrNullViolation, "NOT NULL constraint violation for field %s\n", def->fldName_);
 802             return ErrNullViolation;
 803         }
 804         if (def->isDefault_ && NULL == def->bindVal_ && isInsert)
 805         {
 806             void *dest = AllDataType::alloc(def->type_, def->length_);
 807             AllDataType::convert(typeString, def->defaultValueBuf_, def->type_, dest, def->length_);
 808             AllDataType::copyVal(colPtr, dest, def->type_, def->length_);
 809             colPtr = colPtr + def->length_;
 810             fldpos++;
 811             free (dest);
 812             continue;
 813         }
 814         switch(def->type_)
 815         {
 816             case typeString:
 817                 if (NULL != def->bindVal_)
 818                 {
 819                     if(!isInsert && isFldNull(fldpos)){clearNullBit(fldpos);}
 820                     strcpy((char*)colPtr, (char*)def->bindVal_);
 821                     *(((char*)colPtr) + (def->length_-1)) = '\0';
 822                 }
 823                 else if (!def->isNull_ && !def->bindVal_ && isInsert)  setNullBit(fldpos);
 824                 colPtr = colPtr + def->length_;
 825                 break;
 826             case typeBinary:
 827                 if (NULL != def->bindVal_ )
 828                 {
 829                     if(!isInsert && isFldNull(fldpos)){clearNullBit(fldpos);}
 830                     DbRetVal rv = AllDataType::strToValue(colPtr, (char *) def->bindVal_, def->type_, def->length_);
 831                     if (rv != OK) return ErrBadArg;
 832                 }
 833                 else if (!def->isNull_ && isInsert && !def->bindVal_)  setNullBit(fldpos);
 834                 colPtr = colPtr + def->length_;
 835                 break;
 836             default:
 837                 if (NULL != def->bindVal_){
 838                     if(!isInsert && isFldNull(fldpos)){clearNullBit(fldpos);}
 839                     AllDataType::copyVal(colPtr, def->bindVal_, def->type_);}
 840                 else { if (!def->isNull_ && isInsert)  setNullBit(fldpos); }
 841                 colPtr = colPtr + def->length_;
 842                 break;
 843         }
 844         fldpos++;
 845     }
 846     return OK;
 847 }
 848 void TableImpl::clearNullBit(int fldpos)
 849 {
 850     if (isIntUsedForNULL){
 851         CLEARBIT(iNullInfo, fldpos);}
 852     else
 853         cNullInfo[fldpos-1] = 0;
 854 }
 855 void TableImpl::setNullBit(int fldpos)
 856 {
 857     if (isIntUsedForNULL)
 858         SETBIT(iNullInfo, fldpos);
 859     else
 860         cNullInfo[fldpos-1] = 1;
 861 }
 862 DbRetVal TableImpl::copyValuesToBindBuffer(void *tuplePtr)
 863 {
 864     //Iterate through the bind list and copy the value here
 865     char *colPtr = (char*) tuplePtr;
 866     FieldDef *def = NULL;
 867     for (int i = 0; i < numBindFlds_; i++) {
 868        def = (FieldDef *) bindListArray_[i];
 869        colPtr =  (char *) tuplePtr + def->offset_;
 870        AllDataType::copyVal(def->bindVal_, colPtr, def->type_, def->length_);
 871     }
 872     return OK;
 873 }
 874
 875 //-1 index not supported
 876 DbRetVal TableImpl::insertIndexNode(Transaction *tr, void *indexPtr, IndexInfo *info, void *tuple)
 877 {
 878     CINDEX *iptr = (CINDEX*)indexPtr;
 879     DbRetVal ret = OK;
 880     printDebug(DM_Table, "Inside insertIndexNode type %d", iptr->indexType_);
 881     Index* idx = Index::getIndex(iptr->indexType_);
 882     ret = idx->insert(this, tr, indexPtr, info, tuple,undoFlag);
 883     return ret;
 884 }
 885
 886 DbRetVal TableImpl::deleteIndexNode(Transaction *tr, void *indexPtr, IndexInfo *info, void *tuple)
 887 {
 888     CINDEX *iptr = (CINDEX*)indexPtr;
 889     DbRetVal ret = OK;
 890     Index* idx = Index::getIndex(iptr->indexType_);
 891     ret = idx->remove(this, tr, indexPtr, info, tuple, undoFlag);
 892     return ret;
 893 }
 894 void TableImpl::printSQLIndexString()
 895 {
 896     CatalogTableINDEXFIELD cIndexField(sysDB_);
 897     char fName[IDENTIFIER_LENGTH];
 898     char *fldName = fName;
 899     DataType type;
 900     for (int i = 0; i < numIndexes_ ; i++)
 901     {
 902         CINDEX *iptr = (CINDEX*) indexPtr_[i];
 903         printf("CREATE INDEX %s on %s ( ", iptr->indName_, getName());
 904         FieldList fldList;
 905         cIndexField.getFieldInfo(iptr, fldList);
 906         FieldIterator fIter = fldList.getIterator();
 907         bool firstFld = true;
 908         while(fIter.hasElement())
 909         {
 910             FieldDef *def = fIter.nextElement();
 911             if (firstFld) { printf(" %s ", def->fldName_); firstFld = false; }
 912             else printf(" ,%s ", def->fldName_);
 913         }
 914         printf(" ) ");
 915         if (iptr->indexType_ == hashIndex) printf(" HASH ");
 916         else printf(" TREE ");
 917         if (((HashIndexInfo*) idxInfo[i])->isUnique) printf(" UNIQUE;\n"); else printf(";\n");
 918     }
 919 }
 920
 921
 922 DbRetVal TableImpl::updateIndexNode(Transaction *tr, void *indexPtr, IndexInfo *info, void *tuple)
 923 {
 924     CINDEX *iptr = (CINDEX*)indexPtr;
 925     DbRetVal ret = OK;
 926     Index* idx = Index::getIndex(iptr->indexType_);
 927     //TODO::currently it updates irrespective of whether the key changed or not
 928     //because of this commenting the whole index update code. relook at it and uncomment
 929
 930     ret = idx->update(this, tr, indexPtr, info, tuple, undoFlag);
 931
 932     return ret;
 933 }
 934
 935
 936 void TableImpl::setTableInfo(char *name, int tblid, size_t  length,
 937                        int numFld, int numIdx, void *chunk)
 938 {
 939     strcpy(tblName_, name);
 940     tblID_ = tblid;
 941     length_ = length;
 942     numFlds_ = numFld;
 943     numIndexes_ = numIdx;
 944     chunkPtr_ = chunk;
 945 }
 946
 947 long TableImpl::spaceUsed()
 948 {
 949     Chunk *chk = (Chunk*)chunkPtr_;
 950     long totSize = chk->getTotalDataNodes() * chk->getSize();
 951     totSize = totSize + (chk->totalPages() * sizeof (PageInfo));
 952     return totSize;
 953 }
 954
 955 int TableImpl::pagesUsed()
 956 {
 957     Chunk *chk = (Chunk*)chunkPtr_;
 958     return chk->totalPages();
 959 }
 960
 961 long TableImpl::numTuples()
 962 {
 963     return ((Chunk*)chunkPtr_)->getTotalDataNodes();
 964 }
 965
 966 List TableImpl::getFieldNameList()
 967 {
 968     List fldNameList;
 969     FieldIterator fIter = fldList_.getIterator();
 970     char fieldName[IDENTIFIER_LENGTH];
 971     while (fIter.hasElement())
 972     {
 973         FieldDef *def = fIter.nextElement();
 974         Identifier *elem = new Identifier();
 975         Table::getFieldNameAlone(def->fldName_, fieldName);
 976         sprintf(elem->name, "%s.%s", getName(), fieldName);
 977         fldNameList.append(elem);
 978     }
 979     return fldNameList;
 980 }
 981 DbRetVal TableImpl::close()
 982 {
 983     if (NULL == iter)
 984     {
 985          //printError(ErrNotOpen,"Scan not open");
 986          //return ErrNotOpen;
 987          //PRABA::when called multiple times it gives error
 988          return OK;
 989     }
 990     iter->close();
 991     delete iter;
 992     iter = NULL;
 993     return OK;
 994 }
 995 DbRetVal TableImpl::closeScan()
 996 {
 997     //do not throw scan not open error
 998     //this function will be called by table handle
 999     if (iter) {
1000         iter->reset();
1001         //PRABA::TEMP::otherwise fails.check with kishor
1002         delete iter;
1003         iter = NULL;
1004     }
1005     return OK;
1006 }
1007 DbRetVal TableImpl::lock(bool shared)
1008 {
1009
1010     DbRetVal ret = OK;
1011 /*
1012     if (shared)
1013         ret = lMgr_->getSharedLock(chunkPtr_, NULL);
1014     else
1015         ret = lMgr_->getExclusiveLock(chunkPtr_, NULL);
1016     if (OK != ret)
1017     {
1018         printError(ret, "Could not exclusive lock on the table %x", chunkPtr_);
1019     }else {
1020         //do not append for S to X upgrade
1021         if (!ProcessManager::hasLockList.exists(chunkPtr_))
1022             ProcessManager::hasLockList.append(chunkPtr_);
1023     }
1024 */
1025     return ret;
1026 }
1027 DbRetVal TableImpl::unlock()
1028 {
1029 /*
1030     if (!ProcessManager::hasLockList.exists(chunkPtr_)) return OK;
1031     DbRetVal ret = lMgr_->releaseLock(chunkPtr_);
1032     if (OK != ret)
1033     {
1034         printError(ret, "Could not release exclusive lock on the table %x", chunkPtr_);
1035     }else
1036     {
1037         ProcessManager::hasLockList.remove(chunkPtr_);
1038     }
1039 */
1040     return OK;
1041 }
1042
1043 TableImpl::~TableImpl()
1044 {
1045     if (NULL != iter ) { delete iter; iter = NULL; }
1046     if (NULL != indexPtr_) { delete[] indexPtr_; indexPtr_ = NULL;  }
1047     if (NULL != idxInfo)
1048     {
1049         for (int i = 0; i < numIndexes_; i++) delete idxInfo[i];
1050         delete[] idxInfo;
1051         idxInfo = NULL;
1052     }
1053     if (numFlds_ > 31 && cNullInfo != NULL) { free(cNullInfo); cNullInfo = NULL; }
1054
1055     fldList_.removeAll();
1056
1057 }
1058
1059 void *TableImpl::getBindFldAddr(const char *name)
1060 {
1061         return fldList_.getBindField(name);
1062 }
1063 bool TableImpl::isTableInvolved(char  *tblName)
1064 {
1065     //printf("Table isTableInvolved called for %s with %s\n", tblName, getName());
1066     if (0 == strcmp(getName(), tblName)) return true; else return false;
1067 }
1068 bool TableImpl::pushPredicate(Predicate *pred)
1069 {
1070     bool ret = false;
1071     PredicateImpl *pImpl = (PredicateImpl*) pred;
1072     char tableName[IDENTIFIER_LENGTH];
1073     Table::getTableNameAlone(pImpl->getFldName1(), tableName);
1074     //printf("predicate tbl name %s\n", tableName);
1075
1076     //if predicate is of form t1.f1=t2.f1 then do not push here
1077     if (0 != strcmp(pImpl->getFldName2(),"")) return ret;
1078
1079     if (0 == strcmp(getName(), tableName))
1080     {
1081         setPredicate(pred);
1082         //printf("PRABA::pushed predicate in tablehdl  %s\n", getName());
1083         ret = true;
1084     }
1085     return ret;
1086 }
1087 void TableImpl::setPredicate(Predicate *pred)
1088 {
1089     if (NULL == pred_) { pred_ = pred; return; }
1090     Predicate *curPred = pred_;
1091     PredicateImpl *newPred = new PredicateImpl();
1092     newPred->setTerm(curPred, OpAnd, pred);
1093     newPred->setTable(this);
1094     pred_ = newPred;
1095     return;
1096 }
1097 void TableImpl::printPlan(int space)
1098 {
1099     char spaceBuf[IDENTIFIER_LENGTH];
1100     memset(spaceBuf, 32, IDENTIFIER_LENGTH);
1101     spaceBuf[space] = '\0';
1102     printf("%s <TABLE-NODE>\n", spaceBuf);
1103     printf("%s <NAME> %s </NAME>\n", spaceBuf, getName());
1104     printf("%s <ScanType> %s </ScanType>\n", spaceBuf, ScanTypeNames[scanType_]);
1105     PredicateImpl *pred = (PredicateImpl*)pred_;
1106     if (pred) pred->print(space+2);
1107     printf("%s </TABLE-NODE>\n", spaceBuf);
1108 }
1109