src/storage/DbMgrTableImpl.cxx

   1 /***************************************************************************
   2  *   Copyright (C) 2007 by www.databasecache.com                           *
   3  *   Contact: praba_tuty@databasecache.com                                 *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) any later version.                                   *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15   ***************************************************************************/
  16 #include<Database.h>
  17 #include<DatabaseManager.h>
  18 #include<DatabaseManagerImpl.h>
  19 #include<os.h>
  20 #include<Table.h>
  21 #include<TableImpl.h>
  22 #include<Transaction.h>
  23 #include<CatalogTables.h>
  24 #include<Index.h>
  25 #include<Lock.h>
  26 #include<Debug.h>
  27 #include<Config.h>
  28 #include<Process.h>
  29
  30 //-1 -> Unable to create chunk. No memory
  31 //-2 -> Unable to update the catalog tables
  32 DbRetVal DatabaseManagerImpl::createTable(const char *name, TableDef &def)
  33 {
  34     DbRetVal rv = OK;
  35     if (!Util::isIdentifier((char*)name)) {
  36         printError(ErrBadArg, "Invalid character for index name");
  37         return ErrBadArg;
  38     }
  39     int fldCount = def.getFieldCount();
  40     if(0==fldCount)
  41     {
  42         printError(ErrNotExists,"Table can't be created without Field");
  43         return ErrNotExists;
  44     }
  45
  46     //If total field count is within 32, then 1 integer is used to store all
  47     // null information, if it is more then 1 char is used to store null
  48     // information of each field
  49     //This is to done to reduce cpu cycles for small tables
  50     int addSize = 0;
  51     if (fldCount <= 32) addSize = 4; else addSize = os::align(fldCount);
  52     size_t sizeofTuple = os::alignLong(def.getTupleSize()+addSize);
  53     rv = systemDatabase_->getXCheckpointMutex();
  54     if (OK != rv ) {
  55         printError(rv, "Unable to get Database mutex");
  56         return rv;
  57     }
  58
  59     void *tptr =NULL;
  60     void *chunk = NULL;
  61     void *vcchunk = NULL;
  62
  63     //check whether table already exists
  64     CatalogTableTABLE cTable(systemDatabase_);
  65     cTable.getChunkAndTblPtr(name, chunk, tptr, vcchunk);
  66     if (NULL != tptr)
  67     {
  68         systemDatabase_->releaseCheckpointMutex();
  69         printError(ErrAlready, "Table %s already exists", name);
  70         return ErrAlready;
  71     }
  72
  73     //create a chunk to store the tuples
  74     Chunk *ptr = createUserChunk(sizeofTuple);
  75     if (NULL  == ptr)
  76     {
  77         systemDatabase_->releaseCheckpointMutex();
  78         printError(ErrNoResource, "Unable to create user chunk");
  79         return ErrNoResource;
  80     }
  81     printDebug(DM_Database,"Created UserChunk:%x", ptr);
  82     ptr->setChunkName(name);
  83     //add row to TABLE
  84     int tblID = ((Chunk*)ptr)->getChunkID();
  85
  86     //check whether varchar is present in table
  87     FieldIterator fiter = def.getFieldIterator();
  88     bool isVarcharPresent = def.isVarcharPresentInSchema(fiter);
  89     Chunk *vcptr = NULL;
  90     if (isVarcharPresent) {
  91         //creat chunk to store varchar values
  92         vcptr = createUserChunk();
  93         if (NULL  == vcptr)
  94         {
  95             deleteUserChunk(ptr);
  96             systemDatabase_->releaseCheckpointMutex();
  97             printError(ErrNoResource, "Unable to create user chunk for varchar");
  98             return ErrNoResource;
  99         }
 100         printDebug(DM_Database,"Created UserChunk for Varchar:%x", vcptr);
 101         vcptr->setChunkName(name);
 102     }
 103     rv = cTable.insert(name, tblID, sizeofTuple,
 104                                    def.getFieldCount(), ptr, tptr, vcptr);
 105     if (OK != rv)
 106     {
 107         deleteUserChunk(ptr);
 108         if (vcptr) deleteUserChunk(vcptr);
 109         systemDatabase_->releaseCheckpointMutex();
 110         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 111         return ErrSysInternal;
 112     }
 113     printDebug(DM_Database,"Inserted into TABLE:%s",name);
 114     //add rows to FIELD
 115     FieldIterator iter = def.getFieldIterator();
 116     CatalogTableFIELD cField(systemDatabase_);
 117     rv = cField.insert(iter, tblID ,tptr);
 118     if (OK != rv)
 119     {
 120         deleteUserChunk(ptr);
 121         if (vcptr) deleteUserChunk(vcptr);
 122         void *cptr, *ttptr;//Dummy as remove below needs both these OUT params
 123         cTable.remove(name, cptr, ttptr);
 124         systemDatabase_->releaseCheckpointMutex();
 125         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 126         return ErrSysInternal;
 127     }
 128     printDebug(DM_Database,"Inserted into FIELD:%s",name);
 129     systemDatabase_->releaseCheckpointMutex();
 130     printDebug(DM_Database,"Table Created:%s",name);
 131     logFinest(Conf::logger, "Table Created %s" , name);
 132     return OK;
 133 }
 134 DbRetVal DatabaseManagerImpl::renameTable(const char *oldName,const char *newName)
 135 {
 136     void *chunk = NULL;
 137     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 138     if (OK != rv) {
 139         printError(ErrSysInternal, "Unable to get database mutex for rename table");
 140         return ErrSysInternal;
 141     }
 142     CatalogTableTABLE cTable(systemDatabase_);
 143     rv = cTable.renameTable(oldName,newName);
 144     if (OK != rv) {
 145         printError(ErrSysInternal, "Unable to rename table");
 146         systemDatabase_->releaseCheckpointMutex();
 147         return ErrSysInternal;
 148     }
 149     systemDatabase_->releaseCheckpointMutex();
 150     return OK;
 151 }
 152
 153 DbRetVal DatabaseManagerImpl::renameField(const char *tableName,const char *oldName,const char *newName)
 154 {
 155    DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 156     if (OK != rv) {
 157         printError(ErrSysInternal, "Unable to get database mutex for rename table");
 158         return ErrSysInternal;
 159     }
 160     CatalogTableFIELD fTable(systemDatabase_);
 161     rv = fTable.renameField(tableName, oldName, newName);
 162     if (OK != rv) {
 163         printError(ErrSysInternal, "Unable to rename field.");
 164         systemDatabase_->releaseCheckpointMutex();
 165         return ErrSysInternal;
 166     }
 167     systemDatabase_->releaseCheckpointMutex();
 168     return rv;
 169 }
 170
 171 //TODO::If any operation fails in between, then we may have some
 172 //dangling tuples, say we have have rows in INDEX table
 173 //which will not have any corresponding entries in TABLE
 174 //CHANGE the sequence so that it deletes from the bottom as
 175 //opposed to start from top as is written now
 176 DbRetVal DatabaseManagerImpl::dropTable(const char *name)
 177 {
 178     void *chunk = NULL;
 179     void *tptr =NULL;
 180     void *vcchunk = NULL;
 181     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 182     if (OK != rv) {
 183         printError(ErrSysInternal, "Unable to get database mutex");
 184         return ErrSysInternal;
 185     }
 186     //remove the entry in TABLE
 187     CatalogTableTABLE cTable(systemDatabase_);
 188     rv = cTable.getChunkAndTblPtr(name, chunk, tptr, vcchunk);
 189     if (OK != rv) {
 190         systemDatabase_->releaseCheckpointMutex();
 191         printError(ErrSysInternal, "Table %s does not exist", name);
 192         return ErrSysInternal;
 193     }
 194     CatalogTableFK cFK(systemDatabase_);
 195     int noOfRelation =cFK.getNumFkTable(tptr);
 196     if(noOfRelation)
 197     {
 198         printError(ErrSysInternal, "Unable to drop table due to relation exist.Drop child table...");
 199         systemDatabase_->releaseCheckpointMutex();
 200         return ErrSysInternal;
 201     }
 202     txnMgr()->rollback(lockMgr());
 203     txnMgr()->startTransaction(lockMgr(), READ_COMMITTED);
 204     Transaction **trans = ProcessManager::getThreadTransAddr(systemDatabase_->procSlot);
 205     rv = lMgr_->getExclusiveLock(chunk, trans);
 206     if (rv !=OK)
 207     {
 208         systemDatabase_->releaseCheckpointMutex();
 209         txnMgr()->rollback(lockMgr());
 210         printError(ErrLockTimeOut, "Unable to acquire exclusive lock on the table\n");
 211         return rv;
 212     }
 213     rv = cTable.remove(name, chunk, tptr);
 214     if (OK != rv) {
 215         systemDatabase_->releaseCheckpointMutex();
 216         txnMgr()->rollback(lockMgr());
 217         printError(ErrSysInternal, "Unable to update catalog table TABLE");
 218         return ErrSysInternal;
 219     }
 220     printDebug(DM_Database,"Deleted from TABLE:%s",name);
 221
 222     //remove the entries in the FIELD table
 223     CatalogTableFIELD cField(systemDatabase_);
 224     rv = cField.remove(tptr);
 225     if (OK != rv) {
 226         systemDatabase_->releaseCheckpointMutex();
 227         txnMgr()->rollback(lockMgr());
 228         printError(ErrSysInternal, "Unable to update catalog table FIELD");
 229         return ErrSysInternal;
 230     }
 231     printDebug(DM_Database,"Deleted from FIELD:%s",name);
 232
 233     rv = deleteUserChunk((Chunk*)chunk);
 234     if (OK != rv) {
 235         systemDatabase_->releaseCheckpointMutex();
 236         txnMgr()->rollback(lockMgr());
 237         printError(rv, "Unable to delete the chunk");
 238         return rv;
 239     }
 240     printDebug(DM_Database,"Deleted UserChunk:%x", chunk);
 241
 242     if (vcchunk != NULL) {
 243         rv = deleteUserChunk((Chunk*)vcchunk);
 244         if (OK != rv) {
 245             systemDatabase_->releaseCheckpointMutex();
 246             txnMgr()->rollback(lockMgr());
 247             printError(rv, "Unable to delete the chunk");
 248             return rv;
 249         }
 250         printDebug(DM_Database,"Deleted UserChunk for Varchar:%x", chunk);
 251     }
 252
 253     //TODO::check whether indexes are available and drop that also.
 254     CatalogTableINDEX cIndex(systemDatabase_);
 255     int noIndexes = cIndex.getNumIndexes(tptr);
 256     for (int i =1 ; i<= noIndexes; i++) {
 257         char *idxName = cIndex.getIndexName(tptr, 1);
 258         dropIndexInt(idxName, false);
 259     }
 260     bool isFkExist=cFK.isFkTable(tptr);
 261     if(isFkExist)
 262     {
 263        dropForeignKey(tptr,false);
 264     }
 265     systemDatabase_->releaseCheckpointMutex();
 266     printDebug(DM_Database, "Deleted Table %s" , name);
 267     logFinest(Conf::logger, "Deleted Table %s" , name);
 268     rv = txnMgr()->commit(lockMgr());
 269     if (rv !=OK)
 270     {
 271         printError(ErrLockTimeOut, "Unable to release exclusive lock on the table\n");
 272         return rv;
 273     }
 274     return OK;
 275 }
 276
 277 //Return values: NULL for table not found
 278 Table* DatabaseManagerImpl::openTable(const char *name,bool checkpkfk)
 279 {
 280     DbRetVal ret = OK;
 281     //TODO::store table handles in list so that if it is
 282     //not closed by the application. destructor shall close it.
 283     TableImpl *table = new TableImpl();
 284     table->setDB(db_);
 285     table->setSystemDB(systemDatabase_);
 286     table->setLockManager(lMgr_);
 287     table->setTrans(ProcessManager::getThreadTransAddr(systemDatabase_->procSlot));
 288
 289     //to store the chunk pointer of table
 290     void *chunk = NULL;
 291     void *vcchunk = NULL;
 292
 293     //to store the tuple pointer of the table
 294     void *tptr =NULL;
 295
 296     //TODO::need to take shared lock on the table so that
 297     //all ddl operation will be denied on that table
 298     //which includes index creation, alter table
 299
 300     DbRetVal rv = systemDatabase_->getAllocDatabaseMutex();
 301     if (OK != rv) {
 302         printError(ErrSysInternal, "Unable to get database mutex");
 303         delete table;
 304         return NULL;
 305     }
 306     CatalogTableTABLE cTable(systemDatabase_);
 307     ret = cTable.getChunkAndTblPtr(name, chunk, tptr, vcchunk);
 308     if ( OK != ret)
 309     {
 310         systemDatabase_->releaseAllocDatabaseMutex();
 311         delete table;
 312         printError(ErrNotExists, "Table not exists %s", name);
 313         return NULL;
 314     }
 315     CTABLE *tTuple = (CTABLE*)tptr;
 316     table->setTableInfo(tTuple->tblName_, tTuple->tblID_, tTuple->length_,
 317                         tTuple->numFlds_, tTuple->numIndexes_,
 318                         tTuple->chunkPtr_, tTuple->varcharChunkPtr_);
 319     rv = table->lock(true); //take shared lock
 320     if (rv !=OK)
 321     {
 322         printError(ErrLockTimeOut, "Unable to acquire shared lock on the table\n");
 323         systemDatabase_->releaseAllocDatabaseMutex();
 324         delete table;
 325         return NULL;
 326     }
 327
 328     if (tTuple->numFlds_ <= 32)
 329     {
 330         table->isIntUsedForNULL = true;
 331         table->iNullInfo = 0;
 332         table->iNotNullInfo =0;
 333     }
 334     else
 335     {
 336         table->isIntUsedForNULL = false;
 337         int noFields = os::align(tTuple->numFlds_);
 338         table->cNullInfo = (char*) malloc(noFields);
 339         table->cNotNullInfo = (char*) malloc(noFields);
 340         for (int i =0 ; i < noFields; i++) table->cNullInfo[i] =0;
 341         for (int i =0 ; i < noFields; i++) table->cNotNullInfo[i] =0;
 342
 343     }
 344
 345     //get field information from FIELD table
 346     CatalogTableFIELD cField(systemDatabase_);
 347     table->ptrToAuto = cField.getFieldInfo(tptr, table->fldList_);
 348
 349     //populate the notnull info
 350     FieldIterator fIter = table->fldList_.getIterator();
 351     int fldpos=1;
 352     while (fIter.hasElement())
 353     {
 354         FieldDef *def = fIter.nextElement();
 355         if (table->isIntUsedForNULL) {
 356             if (def->isNull_) SETBIT(table->iNotNullInfo, fldpos-1);
 357         }
 358         else {
 359             if (def->isNull_) table->cNotNullInfo[fldpos-1] = 1;
 360         }
 361         fldpos++;
 362    }
 363
 364     //get the number of indexes on this table
 365     //and populate the indexPtr array
 366     CatalogTableINDEX cIndex(systemDatabase_);
 367     table->numIndexes_ = cIndex.getNumIndexes(tptr);
 368     if (table->numIndexes_) {
 369         table->indexPtr_ = new char*[table->numIndexes_];
 370         table->idxInfo = new IndexInfo*[table->numIndexes_];
 371     }
 372     else
 373     {
 374         table->indexPtr_ = NULL;
 375     }
 376     cIndex.getIndexPtrs(tptr, table->indexPtr_);
 377     for (int i =0 ; i < table->numIndexes_; i++ )
 378     {
 379         HashIndexInfo *hIdxInfo = new HashIndexInfo();
 380         CatalogTableINDEXFIELD cIndexField(systemDatabase_);
 381         cIndexField.getFieldInfo(table->indexPtr_[i], hIdxInfo->idxFldList);
 382         ChunkIterator citer = CatalogTableINDEX::getIterator(table->indexPtr_[i]);
 383         hIdxInfo->indexPtr = table->indexPtr_[i];
 384         hIdxInfo->indType = ((CINDEX*)hIdxInfo->indexPtr)->indexType_;
 385         hIdxInfo->noOfBuckets = CatalogTableINDEX::getNoOfBuckets(table->indexPtr_[i]);
 386         FieldIterator fIter = hIdxInfo->idxFldList.getIterator();
 387         bool firstFld = true;
 388         while (fIter.hasElement())
 389         {
 390             FieldDef *def = fIter.nextElement();
 391             if (firstFld)
 392             {
 393                 hIdxInfo->fldOffset = table->fldList_.getFieldOffset(def->fldName_);
 394                 hIdxInfo->type = table->fldList_.getFieldType(def->fldName_);
 395                 hIdxInfo->compLength = table->fldList_.getFieldLength(def->fldName_);
 396                 firstFld = false;
 397             }else {
 398                 hIdxInfo->type = typeComposite;
 399                 hIdxInfo->compLength = hIdxInfo->compLength +
 400                          table->fldList_.getFieldLength(def->fldName_);
 401             }
 402         }
 403
 404         hIdxInfo->isUnique = CatalogTableINDEX::getUnique(table->indexPtr_[i]);
 405         hIdxInfo->buckets = (Bucket*)citer.nextElement();
 406         table->idxInfo[i] = (IndexInfo*) hIdxInfo;
 407     }
 408     systemDatabase_->releaseAllocDatabaseMutex();
 409     //Foreign key Operation
 410     if(checkpkfk){
 411         CatalogTableFK cFk(systemDatabase_);
 412         int totalFld=0;
 413         table->numFkRelation_ = cFk.getNumFkTable(tptr);
 414         if (table->numFkRelation_) {
 415             table->isPkTbl=true;//TODO:for Delete In casecade
 416             totalFld=cFk.getNoOfFkTable(tptr);
 417             //printDebug(DM_TEST,"Total table is %d\n",totalFld);
 418             char **fptr = new char* [totalFld];
 419             cFk.getFkTableName(tptr,fptr);
 420             for(int count=0; count < totalFld; count++){
 421                //printDebug(DM_TEST,"FK Name is %s\n",fptr[count]);
 422                Table *pkTable=openTable(fptr[count],false);
 423                if (pkTable) table->tblFkList.append(pkTable);
 424                else {
 425                    printError(ErrSysInternal, "Unable to open foreign key tables");
 426                    delete[] fptr;
 427                    pkTable->close();
 428                    return NULL;
 429                }
 430             }
 431             delete[] fptr;
 432
 433         }
 434         char *tblName = NULL;
 435         table->isFkTbl = cFk.isFkTable(tptr);
 436         if(table->isFkTbl)
 437         {
 438             totalFld=cFk.getNoOfPkTable(tptr);
 439             char **fptr = new char* [totalFld];
 440             cFk.getPkTableName(tptr,fptr);
 441             for(int count=0; count<totalFld; count++){
 442                 //printDebug(DM_TEST,"Parent Name is %s\n",fptr[count]);
 443                 Table *fkTable = openTable(fptr[count],false);
 444                 if (fkTable) table->tblList.append(fkTable);
 445                 else {
 446                    printError(ErrSysInternal, "Unable to open foreign key tables");
 447                    delete[] fptr;
 448                    fkTable->close();
 449                    return NULL;
 450                }
 451             }
 452             delete[] fptr;
 453         }
 454     }
 455     printDebug(DM_Database,"Opening table handle name:%s chunk:%x numIndex:%d",
 456                                          name, chunk, table->numIndexes_);
 457     logFinest(Conf::logger, "Opening Table %s" , name);
 458     return table;
 459 }
 460
 461 List DatabaseManagerImpl::getAllTableNames(int *retval)
 462 {
 463     DbRetVal ret = OK;
 464     //to store the tuple pointer of the table
 465     void *tptr =NULL;
 466
 467     /*DbRetVal rv = systemDatabase_->getSCheckpointMutex();
 468     if (OK != rv) {
 469         printError(ErrSysInternal, "Unable to get checkpoint mutex");
 470         if(retval) *retval = rv;
 471         List tableList;
 472         return tableList;
 473     }*/
 474     CatalogTableTABLE cTable(systemDatabase_);
 475     List tableList = cTable.getTableList();
 476     //systemDatabase_->releaseCheckpointMutex();
 477     return tableList;
 478 }
 479
 480
 481 //Return values: -1 for table not found
 482 void DatabaseManagerImpl::closeTable(Table *table)
 483 {
 484     printDebug(DM_Database,"Closing table handle: %x", table);
 485     if (NULL == table) return;
 486     table->unlock();
 487  /*   TableImpl *fkTbl =NULL;
 488     ListIterator tblIter = ((TableImpl*)table)->tblList.getIterator();
 489     tblIter.reset();
 490     while (tblIter.hasElement()){
 491         fkTbl = (TableImpl *) tblIter.nextElement();
 492         closeTable(fkTbl);
 493     }
 494     ((TableImpl*)table)->tblList.reset();
 495     tblIter = ((TableImpl*)table)->tblFkList.getIterator();
 496     tblIter.reset();
 497     while (tblIter.hasElement()){
 498         fkTbl = (TableImpl *) tblIter.nextElement();
 499         closeTable(fkTbl);
 500     }
 501     ((TableImpl*)table)->tblFkList.reset();*/
 502     if (table) delete table; table = NULL;
 503     logFinest(Conf::logger, "Closing Table");
 504 }
 505
 506 int DatabaseManagerImpl::getNoOfPagesForTable(char *tblName)
 507 {
 508     Table *tbl = openTable(tblName);
 509     if (NULL == tbl) {
 510         printError(ErrSysInternal, "Unable to open table %s", tblName);
 511         return 0;
 512     }
 513     TableImpl *tb = (TableImpl *) tbl;
 514     int pages = 0;
 515     if (tb->numTuples()) pages = tb->pagesUsed();
 516     closeTable(tbl);
 517     return pages;
 518 }
 519