src/cache/CacheTableLoader.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<os.h>
  17 #include<CacheTableLoader.h>
  18 #include<TableConfig.h>
  19 #include<Util.h>
  20 #include<SqlConnection.h>
  21 #include<SqlLogConnection.h>
  22 #include<SqlStatement.h>
  23 #include<SqlFactory.h>
  24
  25 DbRetVal CacheTableLoader::load(bool tabDefinition)
  26 {
  27     AbsSqlConnection *conn = SqlFactory::createConnection(CSqlLog);
  28     DbRetVal rv = conn->connect(userName, password);
  29     if (rv != OK) { delete conn; return ErrSysInit; }
  30     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlLog);
  31     stmt->setConnection(conn);
  32     SqlLogConnection *logConn = (SqlLogConnection *) conn;
  33     logConn->setNoMsgLog(true);
  34     logConn->setNoOfflineLog(true);
  35     SqlConnection *con = (SqlConnection *) conn->getInnerConnection();
  36     DatabaseManager *dbMgr = con->getConnObject().getDatabaseManager();
  37     dbMgr->setCanTakeCheckPoint(false);
  38     if (tabDefinition == false) {
  39         rv = hasRecords(dbMgr, tableName);
  40         if (OK != rv) {
  41             dbMgr->setCanTakeCheckPoint(true);
  42             conn->disconnect();
  43             delete stmt;
  44             delete conn;
  45             return ErrNotEmpty;
  46         }
  47     }
  48     conn->beginTrans();
  49     rv = load(conn, stmt, tabDefinition);
  50     conn->commit();
  51     stmt->free();
  52     dbMgr->setCanTakeCheckPoint(true);
  53     conn->disconnect();
  54     delete stmt;
  55     delete conn;
  56     return rv;
  57 }
  58 DbRetVal CacheTableLoader::hasRecords(DatabaseManager *dbMgr, const char* tableName)
  59 {
  60     Table *tbl = dbMgr->openTable(tableName);
  61     if (tbl == NULL) {
  62         printError(ErrNotEmpty, "The table '\%s\' is not found", tableName);
  63         return ErrNotExists;
  64     }
  65     if (tbl->numTuples()) {
  66         printError(ErrNotEmpty, "The table '\%s\' is not empty", tableName);
  67         dbMgr->closeTable(tbl);
  68         return ErrNotEmpty;
  69     }
  70     dbMgr->closeTable(tbl);
  71     return OK;
  72 }
  73 DbRetVal CacheTableLoader::connect(SQLHENV &henv, SQLHDBC &hdbc, SQLHSTMT &hstmt, char *dsn, char* stmtBuf)
  74 {
  75     SQLCHAR outstr[1024];
  76     SQLSMALLINT outstrlen;
  77     int retValue =0;
  78     retValue = SQLAllocHandle (SQL_HANDLE_ENV, SQL_NULL_HANDLE, &henv);
  79     if (retValue) {
  80         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
  81         return ErrSysInit;
  82     }
  83     SQLSetEnvAttr(henv, SQL_ATTR_ODBC_VERSION, (void *) SQL_OV_ODBC3, 0);
  84     retValue = SQLAllocHandle (SQL_HANDLE_DBC, henv, &hdbc);
  85     if (retValue) {
  86         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
  87         return ErrSysInit;
  88     }
  89     retValue = SQLDriverConnect(hdbc, NULL, (SQLCHAR*)dsn, SQL_NTS,
  90                          outstr, sizeof(outstr), &outstrlen,
  91                          SQL_DRIVER_NOPROMPT);
  92     if (SQL_SUCCEEDED(retValue)) {
  93         printDebug(DM_Gateway, "Connected to target database using dsn = %s\n", dsn);
  94     } else {
  95         printError(ErrSysInit, "Failed to connect to target database\n");
  96         return ErrSysInit;
  97     }
  98
  99     retValue=SQLAllocHandle (SQL_HANDLE_STMT, hdbc, &hstmt);
 100     if (retValue) {
 101         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 102         return ErrSysInit;
 103     }
 104
 105     retValue = SQLPrepare (hstmt, (unsigned char *) stmtBuf, SQL_NTS);
 106     if (retValue) {
 107         printError(ErrSysInit, "Unable to Prepare ODBC statement \n");
 108         return ErrSysInit;
 109     }
 110     return OK;
 111 }
 112
 113 DbRetVal CacheTableLoader::load(AbsSqlConnection *conn, AbsSqlStatement *stmt, bool tabDefinition)
 114 {
 115     char dsn[IDENTIFIER_LENGTH];
 116     TDBInfo tdbName = mysql;
 117     DbRetVal rv = OK;
 118
 119     bool isDSNExist = resolveForDSN(dsn, tdbName, rv);
 120     if (!isDSNExist) return rv;
 121
 122     SqlConnection *con = (SqlConnection *) conn->getInnerConnection();
 123     DatabaseManager *dbMgr = con->getConnObject().getDatabaseManager();
 124     SQLHENV henv;
 125     SQLHDBC hdbc;
 126     SQLHSTMT hstmt;
 127     int retValue =0;
 128
 129     /*SQLCHAR outstr[1024];
 130     SQLSMALLINT outstrlen;
 131     SQLHENV henv;
 132     SQLHDBC hdbc;
 133     SQLHSTMT hstmt;
 134     retValue = SQLAllocHandle (SQL_HANDLE_ENV, SQL_NULL_HANDLE, &henv);
 135     if (retValue) {
 136         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 137         return ErrSysInit;
 138     }
 139     SQLSetEnvAttr(henv, SQL_ATTR_ODBC_VERSION, (void *) SQL_OV_ODBC3, 0);
 140     retValue = SQLAllocHandle (SQL_HANDLE_DBC, henv, &hdbc);
 141     if (retValue) {
 142         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 143         return ErrSysInit;
 144     }
 145     retValue = SQLDriverConnect(hdbc, NULL, (SQLCHAR*)dsn, SQL_NTS,
 146                          outstr, sizeof(outstr), &outstrlen,
 147                          SQL_DRIVER_NOPROMPT);
 148     if (SQL_SUCCEEDED(retValue)) {
 149         printDebug(DM_Gateway, "Connected to target database using dsn = %s\n", dsn);
 150     } else {
 151         printError(ErrSysInit, "Failed to connect to target database\n");
 152         return ErrSysInit;
 153     }
 154
 155     retValue=SQLAllocHandle (SQL_HANDLE_STMT, hdbc, &hstmt);
 156     if (retValue) {
 157         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 158         return ErrSysInit;
 159     }
 160
 161     retValue = SQLPrepare (hstmt, (unsigned char *) stmtBuf, SQL_NTS);
 162     if (retValue) {
 163         printError(ErrSysInit, "Unable to Prepare ODBC statement \n");
 164         return ErrSysInit;
 165     }*/
 166     char stmtBuf[1024];
 167     generateCacheTableStatement(stmtBuf);
 168     connect(henv, hdbc, hstmt, dsn, stmtBuf);
 169     int nRecordsToFetch = Conf::config.getNoOfRowsToFetchFromTDB();
 170     int nFetchedRecords = 0;
 171     SQLUSMALLINT *rowStatus = (SQLUSMALLINT *)
 172                                 malloc(nRecordsToFetch * sizeof(SQLUSMALLINT));
 173     memset(rowStatus, 0, nRecordsToFetch * sizeof(SQLUSMALLINT));
 174     SQLSetStmtAttr(hstmt, SQL_ATTR_ROW_BIND_TYPE, SQL_BIND_BY_COLUMN, 0);
 175     SQLSetStmtAttr(hstmt, SQL_ATTR_ROW_ARRAY_SIZE, (void *) nRecordsToFetch, 0);
 176     SQLSetStmtAttr(hstmt, SQL_ATTR_ROW_STATUS_PTR, rowStatus, 0 );
 177     SQLSetStmtAttr(hstmt, SQL_ATTR_ROWS_FETCHED_PTR, &nFetchedRecords, 0 );
 178     if (tabDefinition) {
 179         short totalFields=0;
 180         retValue = SQLNumResultCols (hstmt, &totalFields);
 181         if (retValue) {
 182             printError(ErrSysInit, "Unable to retrieve ODBC total columns\n");
 183             return ErrSysInit;
 184         }
 185         logFinest(Conf::logger, "Cache Table noOfFields %hd", totalFields);
 186         UWORD                   icol=1;
 187         UCHAR                   colName[IDENTIFIER_LENGTH];
 188         SWORD                   colNameMax=0;
 189         SWORD                   nameLength=0;
 190         SWORD                   colType=0;
 191         SQLULEN                 colLength = 0;
 192         SWORD                   scale=0;
 193         SWORD                   nullable=0;
 194         colNameMax = IDENTIFIER_LENGTH;
 195         char columnname[IDENTIFIER_LENGTH];
 196         short type; short unique;
 197         SQLHSTMT hstmtmeta;
 198         retValue=SQLAllocHandle (SQL_HANDLE_STMT, hdbc, &hstmtmeta);
 199         if (retValue)
 200         {
 201             printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 202             return ErrSysInit;
 203         }
 204
 205         char crtIdxStmt[1024];
 206         char *ptr=crtIdxStmt;
 207         HashIndexInitInfo *inf = new HashIndexInitInfo();
 208         bool isPriIndex = prepareCreateIndexStatement(hstmtmeta, crtIdxStmt, tdbName, inf);
 209
 210         bool iskeyfieldExist=false;
 211         bool isPKFieldSpecified = false;
 212         char *name = NULL;
 213         if((strcmp(fieldName,"")!=0) && (strcmp(fieldName,"NULL")!=0) )
 214         {
 215             isPKFieldSpecified = true;
 216         }
 217         if ( isPriIndex && ( strcmp(fieldlistVal,"")!=0 ) &&
 218                                        ( strcmp(fieldlistVal,"NULL") != 0 )) {
 219             inf->list.resetIter();
 220             while ( (name=inf->list.nextFieldName()) != NULL) {
 221                 iskeyfieldExist = TableConf::config.isFieldExist(name);
 222                 if(!iskeyfieldExist) { break; }
 223             }
 224         } else if (isPriIndex) { iskeyfieldExist = true; }
 225         if ( isPKFieldSpecified &&  !(TableConf::config.isFieldExist(fieldName)) )
 226         {
 227             if ( Conf::config.useTwoWayCache() &&
 228                              (strcmp(fieldlistVal,"")!=0) &&
 229                                          (strcmp(fieldlistVal,"NULL")!=0))
 230             {
 231                 printError(ErrSysInit, "Bidirectional caching should have primary key in %s \n", tableName);
 232                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 233                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 234                 SQLDisconnect (hdbc);
 235                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 236                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 237                 delete inf;
 238                 return ErrSysInit;
 239             }
 240         }
 241         if (!iskeyfieldExist && !isPKFieldSpecified )
 242         {
 243             if(Conf::config.useTwoWayCache())
 244             {
 245                 printError(ErrSysInit, "Bidirectional caching fail for no primary key in %s \n", tableName);
 246                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 247                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 248                 SQLDisconnect (hdbc);
 249                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 250                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 251                 delete inf;
 252                 return ErrSysInit;
 253             }
 254         }
 255
 256         bool isKeyFld=false;
 257         char crtTblStmt[1024];
 258         rv = prepareCreateTableStatement(crtTblStmt, hstmt, inf, totalFields,
 259                                                             tdbName, isKeyFld);
 260         if (rv != OK) {
 261             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 262             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 263             SQLDisconnect (hdbc);
 264             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 265             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 266             delete inf;
 267             return ErrSysInit;
 268         }
 269         //printf("table stmt '%s'\n", crtTblStmt);
 270         if(((strcmp(fieldName,"")!=0) && (strcmp(fieldName,"NULL")!=0))
 271                                                                && !isKeyFld) {
 272             printError(ErrSysInit, "Unable to cache Table for %s with key field %s\n", tableName,fieldName);
 273             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 274             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 275             SQLDisconnect (hdbc);
 276             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 277             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 278             delete inf;
 279             return ErrSysInit;
 280         }
 281         rv = stmt->prepare(crtTblStmt);
 282         if (rv != OK) {
 283             printError(ErrSysInit, "Unable to prepare create table stmt\n");
 284             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 285             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 286             SQLDisconnect (hdbc);
 287             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 288             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 289             delete inf;
 290             return ErrSysInit;
 291         }
 292         int rows = 0;
 293         rv = stmt->execute(rows);
 294         if (rv != OK) {
 295             printError(ErrSysInit, "Unable to execute create table stmt\n");
 296             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 297             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 298             SQLDisconnect (hdbc);
 299             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 300             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 301             delete inf;
 302             return ErrSysInit;
 303         }
 304         logFinest(Conf::logger, "Cache Table: Table Created :%s", crtTblStmt);
 305
 306         //Table is created.
 307         //Create primary key index if present
 308         if (isPriIndex && ( iskeyfieldExist ||
 309            (strcmp(fieldlistVal,"")==0  || strcmp(fieldlistVal,"NULL")== 0))) {
 310             rv = stmt->prepare(crtIdxStmt);
 311             if (rv != OK) {
 312                 printError(ErrSysInit, "Unable to prepare create table stmt\n");
 313                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 314                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 315                 SQLDisconnect (hdbc);
 316                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 317                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 318                 delete inf;
 319                 return ErrSysInit;
 320             }
 321             int rows = 0;
 322             rv = stmt->execute(rows);
 323             if (rv != OK) {
 324                 printError(ErrSysInit, "Unable to execute create table stmt\n");
 325                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 326                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 327                 SQLDisconnect (hdbc);
 328                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 329                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 330                 delete inf;
 331                 return ErrSysInit;
 332             }
 333             //printf("Primary index created from create Index stmt\n");
 334         }
 335         retValue = SQLCloseCursor(hstmtmeta);
 336         rv = createIndex(hstmtmeta, tableName, inf, stmt,isPKFieldSpecified);
 337         if(rv!=OK) {
 338             dbMgr->dropTable(tableName);
 339             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 340             SQLDisconnect (hdbc);
 341             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 342             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 343             delete inf;
 344             return rv;
 345         }
 346         logFinest(Conf::logger, "Cache Table: Index :%s", crtIdxStmt);
 347         delete inf;
 348     } // tableDefinition scope finishes here
 349
 350     else { /***Checking for Table Schema between CSQL and TDB(cachetable -s option)***/
 351         rv=checkingSchema(hdbc,hstmt,conn,stmt,tdbName);
 352         if(rv != OK){
 353             printError(ErrSysInit,"Unable to cache the '%s' table due to schema mismatched.",tableName);
 354             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 355             SQLDisconnect (hdbc);
 356             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 357             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 358             return ErrSysInit;
 359         }
 360     } //***Ends Here
 361
 362     // Now load the table with records
 363     stmt->free();
 364
 365     char insStmt[1024];
 366     List fNameList;
 367     fNameList.init();
 368     SqlStatement *sqlStmt = (SqlStatement *)stmt->getInnerStatement();
 369     sqlStmt->setConnection(con);
 370     prepareInsertStatement(sqlStmt, &fNameList, insStmt);
 371     int totalFields = fNameList.size();
 372     rv = stmt->prepare(insStmt);
 373     if (rv != OK) {
 374         printError(ErrSysInit, "Unable to prepare create table stmt\n");
 375         SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 376         SQLDisconnect (hdbc);
 377         SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 378         SQLFreeHandle (SQL_HANDLE_ENV, henv);
 379         return ErrSysInit;
 380     }
 381     sqlStmt->setLoading(true);
 382     ListIterator fNameIter = fNameList.getIterator();
 383     FieldInfo *info = new FieldInfo();
 384     int fcount =1; void *valBuf=NULL;
 385     Identifier *elem = NULL;
 386     void *tembuf=NULL;//For postgre BigInt type
 387     BindBuffer *bBuf;
 388     List valBufList;
 389     int i=0;
 390     while (fNameIter.hasElement()) {
 391         elem = (Identifier*) fNameIter.nextElement();
 392         sqlStmt->getFieldInfo(tableName, (const char*)elem->name, info);
 393         int size = 0;
 394         if (info->type == typeString || info->type == typeVarchar) {
 395             size = nRecordsToFetch * AllDataType::size(info->type,info->length);
 396         } else {
 397             size = nRecordsToFetch * AllDataType::size(info->type);
 398         }
 399         valBuf = malloc(size);
 400         os::memset(valBuf,0,size);
 401         int bindLen = 0;
 402         if (info->type != typeDate && info->type != typeTime && info->type != typeTimeStamp) {
 403             if (info->type == typeLongLong && tdbName == postgres)
 404                 bindLen = 40;
 405             else
 406             bindLen =  AllDataType::size(info->type,
 407                                   AllDataType::size(info->type, info->length));
 408         } else {
 409             switch(info->type) {
 410             case typeDate: bindLen = sizeof(DATE_STRUCT); break;
 411             case typeTime: bindLen = sizeof(TIME_STRUCT); break;
 412             case typeTimeStamp: bindLen = sizeof(TIMESTAMP_STRUCT); break;
 413             }
 414         }
 415
 416         bBuf = (BindBuffer *) SqlStatement::fillBindBuffer(tdbName, info->type, valBuf, bindLen, nRecordsToFetch);
 417         valBufList.append(bBuf);
 418         retValue = SQLBindCol (hstmt, fcount, AllDataType::convertToSQL_C_Type(info->type,tdbName), valBuf, bindLen, bBuf->nullData);
 419         fcount++;
 420         if (retValue) {
 421             if(tabDefinition) dbMgr->dropTable(tableName);
 422             printError(ErrSysInit, "Unable to bind columns in ODBC\n");
 423             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 424             SQLDisconnect (hdbc);
 425             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 426             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 427             return ErrSysInit;
 428         }
 429     }
 430     delete info;
 431     fNameIter.reset();
 432     while (fNameIter.hasElement())
 433         delete ((FieldName *) fNameIter.nextElement());
 434     fNameList.reset();
 435
 436     retValue = SQLExecute (hstmt);
 437     if (retValue) {
 438         printError(ErrSysInit, "Unable to execute ODBC statement\n");
 439         SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 440         SQLDisconnect (hdbc);
 441         SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 442         SQLFreeHandle (SQL_HANDLE_ENV, henv);
 443         return ErrSysInit;
 444     }
 445     int fldpos=0;
 446     do {
 447         //TODO: if SQLFetch return other than record not found error
 448         //it should drop the table
 449         retValue = SQLFetchScroll(hstmt, SQL_FETCH_NEXT, 0);
 450         if (retValue) break;
 451         for (int row = 0; row < nFetchedRecords; row++) {
 452             fldpos = 0;
 453             void *val = NULL;
 454             ListIterator bindIter = valBufList.getIterator();
 455             while (bindIter.hasElement()) {
 456                 bBuf = (BindBuffer*) bindIter.nextElement();
 457                 if (bBuf->nullData[row] == SQL_NULL_DATA) {
 458                     stmt->setNull(fldpos+1);
 459                 } else {
 460                     val = (void *) ((char *)bBuf->csql + row * bBuf->length);
 461                     switch (bBuf->type) {
 462                         case typeString:
 463                         {
 464                             val = (void *) ((char *)bBuf->csql + row * bBuf->length);
 465                             if( tdbName == postgres)
 466                                 Util::trimRight((char*)val);
 467                             break;
 468                         }
 469                         case typeDate:
 470                         {
 471                             val = (void *) ((char *)bBuf->csql + row * sizeof(Date));
 472                             Date *dtCSQL = (Date*) val;
 473                             void *tVal = (void *) ((char *)bBuf->targetdb + row * sizeof(DATE_STRUCT));
 474                             DATE_STRUCT *dtTarget = (DATE_STRUCT*) tVal;
 475                             dtCSQL->set(dtTarget->year,dtTarget->month,dtTarget->day);
 476                             break;
 477                         }
 478                         case typeTime:
 479                         {
 480                             val = (void *) ((char *)bBuf->csql + row * sizeof(Time));
 481                             Time *dtCSQL = (Time*) val;
 482                             void *tVal = (void *) ((char *)bBuf->targetdb + row * sizeof(TIME_STRUCT));
 483                             TIME_STRUCT *dtTarget = (TIME_STRUCT*) tVal;
 484                             dtCSQL->set(dtTarget->hour,dtTarget->minute,dtTarget->second);
 485                             break;
 486                         }
 487                         case typeTimeStamp:
 488                         {
 489                             val = (void *) ((char *)bBuf->csql + row * sizeof(TimeStamp));
 490                             TimeStamp *dtCSQL = (TimeStamp*) val;
 491                             void *tVal = (void *) ((char *)bBuf->targetdb + row * sizeof(TIMESTAMP_STRUCT));
 492                             TIMESTAMP_STRUCT *dtTarget = (TIMESTAMP_STRUCT*) tVal;
 493                             dtCSQL->setDate(dtTarget->year,dtTarget->month,dtTarget->day);
 494                             dtCSQL->setTime(dtTarget->hour,dtTarget->minute,dtTarget->second, dtTarget->fraction);
 495                             break;
 496                         }
 497                         case typeLongLong:
 498                         {
 499                             val = (void *) ((char *)bBuf->csql + row * sizeof(long long));
 500                             void *tVal = (void *) ((char *)bBuf->targetdb + row * bBuf->length);
 501                             if (tdbName == postgres) {
 502                                 sscanf((const char*)tVal,"%lld",(long long*) val);
 503                             }
 504                             break;
 505                         }
 506                     }
 507                     SqlStatement::setParamValues(stmt, fldpos+1, bBuf->type, bBuf->length, val);
 508                 }
 509                 fldpos++;
 510             }
 511             int rows = 0;
 512             rv = stmt->execute(rows);
 513             if (rv != OK) {
 514                 printError(ErrSysInit, "Unable to cache record in CSQL.\n");
 515                 if(tabDefinition) dbMgr->dropTable(tableName);
 516                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 517                 SQLDisconnect (hdbc);
 518                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 519                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 520                 return ErrSysInit;
 521             }
 522             conn->commit();
 523             conn->beginTrans();
 524         }
 525     } while (SQL_SUCCEEDED(retValue) && nFetchedRecords == nRecordsToFetch);
 526     conn->commit();
 527     conn->beginTrans();
 528
 529     //PRABA::one operation per transaction gives the best
 530     //performance than 100 /Txn in case of durability
 531     //TODO::leak:: valBufList and its targetdb buffer
 532     ListIterator it = valBufList.getIterator();
 533     while(it.hasElement()) {
 534         BindBuffer *bb = (BindBuffer *) it.nextElement();
 535         if (bb->csql) { free(bb->csql); bb->csql = NULL; }
 536         if (bb->targetdb) { free(bb->targetdb); bb->targetdb = NULL; }
 537         delete bb; bb = NULL;
 538     }
 539     valBufList.reset();
 540     SQLCloseCursor (hstmt);
 541     SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 542     ::free(rowStatus);
 543     SQLDisconnect (hdbc);
 544     SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 545     SQLFreeHandle (SQL_HANDLE_ENV, henv);
 546     logFine(Conf::logger, "Cached Table: %s", tableName);
 547     return OK;
 548 }
 549
 550 DbRetVal CacheTableLoader::reload()
 551 {
 552     FILE *fp=NULL;
 553     DbRetVal rv = unload(false);
 554     if (rv != OK) return rv;
 555     //get table cache senarios
 556     fp = fopen(Conf::config.getTableConfigFile(),"r");
 557     if( fp == NULL ) {
 558         printError(ErrSysInit, "csqltable.conf file does not exist");
 559         return OK;
 560     }
 561     int mode;
 562     rv = OK;
 563     char tablename[IDENTIFIER_LENGTH];
 564     char fieldname[IDENTIFIER_LENGTH];
 565     char field[IDENTIFIER_LENGTH];
 566     char condition[IDENTIFIER_LENGTH];
 567     char dsnname[IDENTIFIER_LENGTH];
 568     while(!feof(fp))
 569     {
 570         fscanf(fp, "%d %s %s %s %s %s\n", &mode, tablename,fieldname,condition,field,dsnname);
 571         if(strcmp(tablename,tableName)==0) break;
 572     }
 573     fclose(fp);
 574     setCondition(TableConf::config.getRealConditionFromFile(condition));
 575     setFieldName(fieldname);
 576     setFieldListVal(field);
 577     setDsnName(dsnname);
 578     rv = load(false);
 579     return rv;
 580 }
 581
 582 DbRetVal CacheTableLoader::unload(bool tabDefinition)
 583 {
 584     AbsSqlConnection *conn = SqlFactory::createConnection(CSqlLog);
 585     DbRetVal rv = conn->connect(userName, password);
 586     if (rv != OK) return ErrSysInit;
 587     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlLog);
 588     stmt->setConnection(conn);
 589     SqlLogConnection *logConn = (SqlLogConnection *) conn;
 590     logConn->setNoMsgLog(true);
 591     char statement[1024];
 592     if (TableConf::config.isTableCached(tableName) != OK) {
 593         printError(ErrNotCached, "The table \'%s\' is not cached", tableName);
 594         conn->disconnect();
 595         delete stmt;
 596         delete conn;
 597         return ErrNotCached;
 598     }
 599     SqlConnection *con = (SqlConnection *) conn->getInnerConnection();
 600     DatabaseManager *dbMgr = (DatabaseManager*) con->getConnObject().getDatabaseManager();
 601     if (dbMgr == NULL) {
 602         conn->disconnect();
 603         delete stmt; delete conn;
 604         printError(ErrSysInit, "Authentication failed\n");
 605         return ErrSysInit;
 606     }
 607     if (!tabDefinition)
 608     {
 609         sprintf(statement, "DELETE FROM %s;", tableName);
 610         SqlStatement *sqlStmt = (SqlStatement*)stmt;
 611         sqlStmt->setLoading(true);
 612         rv = stmt->prepare(statement);
 613         if (rv != OK) {
 614             conn->disconnect();
 615             delete stmt; delete conn;
 616             return ErrBadCall;
 617         }
 618         conn->beginTrans();
 619         int rows = 0;
 620         rv = stmt->execute(rows);
 621         if (rv != OK) {
 622             conn->disconnect();
 623             delete stmt; delete conn;
 624             return ErrBadCall;
 625         }
 626         conn->commit();
 627     }
 628     else
 629     {
 630         rv = TableConf::config.removeFromCacheTableFile();
 631         if (rv != OK) {
 632             conn->disconnect(); delete stmt; delete conn;
 633             return ErrBadCall;
 634         }
 635         sprintf(statement, "DROP TABLE %s;", tableName);
 636         SqlStatement *sqlStmt = (SqlStatement*)stmt;
 637         sqlStmt->setLoading(true);
 638         rv = stmt->prepare(statement);
 639         if (rv != OK) {
 640             //TableConf::config.addToCacheTableFile(false);
 641             conn->disconnect();
 642             delete stmt; delete conn;
 643             return ErrBadCall;
 644         }
 645         int rows = 0;
 646         rv = stmt->execute(rows);
 647         if (rv != OK) {
 648             //TableConf::config.addToCacheTableFile(false);
 649             conn->disconnect(); delete stmt; delete conn;
 650             return ErrBadCall;
 651         }
 652     }
 653     conn->disconnect();
 654     delete stmt; delete conn;
 655     logFine(Conf::logger, "Unloaded Cached Table: %s", tableName);
 656     return rv;
 657 }
 658
 659 DbRetVal CacheTableLoader::refresh()
 660 {
 661     return OK;
 662 }
 663
 664 DbRetVal CacheTableLoader::recoverAllCachedTables()
 665 {
 666     FILE *fp;
 667     Connection conn;
 668     DbRetVal rv = conn.open(userName, password);
 669     if(rv !=OK) return ErrSysInit;
 670
 671     //Note: if connection is not open, configuration veriables may be incorrect
 672
 673     fp = fopen(Conf::config.getTableConfigFile(),"r");
 674     if( fp == NULL ) {
 675         printError(ErrSysInit, "csqltable.conf file does not exist");
 676         conn.close();
 677         return OK;
 678     }
 679     conn.close();
 680     //TODO::take exclusive lock on database
 681     char tablename[IDENTIFIER_LENGTH];
 682     char fieldname[IDENTIFIER_LENGTH];
 683     char condition[IDENTIFIER_LENGTH];
 684     char field[IDENTIFIER_LENGTH];
 685     char dsnname[IDENTIFIER_LENGTH];
 686
 687     int mode;
 688     int scanItems=0;
 689     rv = OK;
 690     while(!feof(fp))
 691     {
 692         scanItems = fscanf(fp, "%d %s %s %s %s %s\n", &mode, tablename,fieldname,condition,field,dsnname);
 693         if (scanItems != 6) {
 694             tablename[0]='\0';
 695             printf("There is no table to be cached.\n");
 696             return OK;
 697         }
 698         if (!TableConf::config.isTableCached(mode)) continue;
 699         printDebug(DM_Gateway, "Recovering Table from target db: %s\n", tablename);
 700         setCondition(TableConf::config.getRealConditionFromFile(condition));
 701         if( (strcmp(Conf::config.getDSN(),dsnname)!=0) ){
 702                      setDsnName(dsnname);
 703              setTable(tablename);
 704              setFieldName(fieldname);
 705              setFieldListVal(field);
 706              printf("Recovering table %s %s %s\n", tablename,condition,field);
 707              rv = load();
 708              if (rv != OK) { fclose(fp); return rv; }
 709         } else {
 710              setDsnName(Conf::config.getDSN());
 711              setTable(tablename);
 712              setFieldName(fieldname);
 713              setFieldListVal(field);
 714              printf("Recovering table %s %s %s\n", tablename,condition,field);
 715              rv = load();
 716              if (rv != OK) { fclose(fp); return rv; }
 717         }
 718         logFine(Conf::logger, "Recovering Table from target db:%s", tablename);
 719     }
 720     fclose(fp);
 721     return OK;
 722 }
 723
 724 DbRetVal CacheTableLoader::createIndex(SQLHSTMT hstmtmeta, char *tableName, HashIndexInitInfo *inf,AbsSqlStatement *stmt,bool isPKFieldSpecified)
 725 {
 726         bool isKeyFld= false;
 727         int retValue = 0;
 728         char columnname[IDENTIFIER_LENGTH];
 729         char indexname[IDENTIFIER_LENGTH];
 730         short type;
 731         short unique;
 732         char *name = NULL;
 733         DbRetVal rv = OK;
 734         retValue = SQLStatistics(hstmtmeta, NULL, 0, NULL, SQL_NTS,
 735                     (SQLCHAR*) tableName, SQL_NTS, SQL_INDEX_ALL, SQL_QUICK);
 736         retValue = SQLBindCol(hstmtmeta, 4, SQL_C_SHORT,
 737                           &unique, 2, NULL);
 738         retValue = SQLBindCol(hstmtmeta, 6, SQL_C_CHAR,
 739                           indexname, 129, NULL);
 740         retValue = SQLBindCol(hstmtmeta, 7, SQL_C_SHORT,
 741                           &type, 2, NULL);
 742         retValue = SQLBindCol(hstmtmeta, 9, SQL_C_CHAR,
 743                           columnname, 129,NULL);
 744         List indexList;
 745         bool isSecondTime = false;
 746         CacheIndexInfo *info=NULL;
 747         while ((retValue = SQLFetch(hstmtmeta)) == SQL_SUCCESS) {
 748             printDebug(DM_Gateway, "Column: %-18s Index Name: %-18s unique:%hd type:%hd\n", columnname, indexname, unique, type);
 749
 750             if (type == 3) {
 751                 bool isFldAdd = false;
 752                 ListIterator iter = indexList.getIterator();
 753                 iter.reset();
 754                 while (iter.hasElement()) {
 755                     CacheIndexInfo *indInfo = (CacheIndexInfo *)iter.nextElement();
 756                     if(0 == strcmp( indInfo->indexName, indexname))
 757                     {
 758                          indInfo->fieldNameList.append(columnname);
 759                          isFldAdd = true;
 760                     }
 761                 }
 762                 if(!isFldAdd){
 763                     info = new CacheIndexInfo();
 764                     info->fieldNameList.append(columnname);
 765                     strcpy(info->indexName, indexname);
 766                     indexList.append(info);
 767                     isSecondTime = true;
 768                 }
 769             }
 770         }
 771         ListIterator iter = indexList.getIterator();
 772         iter.reset();
 773         int noOfPkfield = inf->list.size();
 774         char *fName=NULL;
 775         char *cptr = NULL;
 776         while (iter.hasElement()) {
 777             cptr = columnname;
 778             bool isFieldExistInCondition = false;
 779             bool isPrimary=false;
 780             CacheIndexInfo *indInfo = (CacheIndexInfo *)iter.nextElement();
 781             int noOfFld= indInfo->fieldNameList.size();
 782             indInfo->fieldNameList.resetIter();
 783             while ((fName = indInfo->fieldNameList.nextFieldName())!=NULL) {
 784                 if(( 1 == noOfFld) && (0 == strcmp(fName,fieldName))) {
 785                     isKeyFld=true;
 786                 }
 787                 inf->list.resetIter();
 788                 while ((name=inf->list.nextFieldName())!=NULL)
 789                 {
 790                     if(0==strcmp(fName,name)) {  isPrimary = true;  break;  }
 791                     isPrimary = false;
 792                 }
 793                         if (!TableConf::config.isFieldExist(fName) &&
 794                                      ( (strcmp(fieldlistVal,"")!=0) &&
 795                                           (strcmp(fieldlistVal,"NULL")!=0) )) {
 796                      isFieldExistInCondition =true;
 797                      continue;
 798                 }
 799                 sprintf(cptr, "%s ,",fName);
 800                 cptr += strlen(cptr);
 801             }
 802             if(isFieldExistInCondition) continue;
 803             cptr -=1;
 804             *cptr = '\0';
 805             if (isPrimary) { continue; }
 806             char crtIdxStmt[1024];
 807             char indname[128];
 808             sprintf(indname, "%s_%s", tableName, indInfo->indexName);
 809             sprintf(crtIdxStmt, "CREATE INDEX %s on %s(%s) HASH SIZE 10007;", indname, tableName, columnname);
 810             //printf("create index stmt \n'%s'\n", crtIdxStmt);
 811             rv = stmt->prepare(crtIdxStmt);
 812             if (rv != OK) {
 813                 printError(ErrSysInit, "Unable to prepare create table stmt\n");
 814                 return ErrSysInit;
 815             }
 816             int rows = 0;
 817             rv = stmt->execute(rows);
 818             if (rv != OK) {
 819                 printError(ErrSysInit, "Unable to execute create table stmt\n");
 820                 return ErrSysInit;
 821             }
 822             //delete indInfo;
 823         }// while meta data fetch for index creation
 824
 825         iter.reset();
 826         while (iter.hasElement()) delete (CacheIndexInfo *) iter.nextElement();
 827         indexList.reset();
 828
 829         SQLCloseCursor (hstmtmeta);
 830         SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 831         if( !isKeyFld && isPKFieldSpecified) {
 832             if(shouldForce) {
 833                 char frcIndStmt[1024];
 834                 char indname[128];
 835                 sprintf(indname, "%s_%s", tableName, "keyInd");
 836                 sprintf(frcIndStmt, "CREATE INDEX %s on %s(%s) HASH;", indname, tableName, fieldName);
 837                 rv = stmt->prepare(frcIndStmt);
 838                 if (rv != OK) {
 839                     printError(ErrSysInit, "Unable to prepare create table stmt\n");
 840                     return ErrSysInit;
 841                 }
 842                 int rows = 0;
 843                 rv = stmt->execute(rows);
 844                 if (rv != OK) {
 845                     printError(ErrSysInit, "Unable to execute create table stmt\n");
 846                     return ErrSysInit;
 847                 }
 848             } else {
 849                 printError(ErrSysInit, "Unable to cache Table for %s with key field %s\n", tableName,fieldName);
 850                 return ErrSysInit;
 851             }
 852         }
 853      return OK;
 854 }
 855
 856 // Schema matching between CSQL and TDB table.(cachetable -t <tablename> -s)
 857 DbRetVal CacheTableLoader::checkingSchema(SQLHDBC hdbc,SQLHSTMT hstmt, AbsSqlConnection *conn, AbsSqlStatement *stmt,TDBInfo tdbName)
 858 {
 859     DbRetVal rv=OK;
 860     int noOfPrimaryKey=0;
 861     int retValue=0;
 862     int csqlFields=0;
 863
 864     SQLSMALLINT tdbFields=0;
 865     SQLHSTMT hstmtmeta;
 866     char columnname[IDENTIFIER_LENGTH];
 867
 868     UWORD                   icol=1;
 869     UCHAR                   colName[IDENTIFIER_LENGTH];
 870     SWORD                   colNameMax=0;
 871     SWORD                   nameLength=0;
 872     SWORD                   colType=0;
 873     SQLULEN                 colLength = 0;
 874     SWORD                   scale=0;
 875     SWORD                   nullable=0;
 876     colNameMax = IDENTIFIER_LENGTH;
 877
 878     SqlConnection *con = (SqlConnection *) conn->getInnerConnection();
 879     DatabaseManager *dbMgr = con->getConnObject().getDatabaseManager();
 880
 881     SqlStatement *sqlStmt = (SqlStatement *)stmt->getInnerStatement();
 882     sqlStmt->setConnection(con);
 883
 884     List fNameList ;
 885     fNameList = sqlStmt->getFieldNameList(tableName, rv);
 886     ListIterator fNameIter = fNameList.getIterator();
 887     FieldInfo *info = new FieldInfo();
 888     Identifier *elem = NULL;
 889
 890     retValue=SQLNumResultCols(hstmt, &tdbFields);
 891     if(retValue) {
 892        printError(ErrSysInit, "Unable to retrieve ODBC total columns.\n");
 893        SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 894        return ErrSysInit;
 895     }
 896     // CSQL Table fields
 897     fNameList = sqlStmt->getFieldNameList(tableName, rv);
 898     csqlFields = fNameList.size();
 899     // noOfFields in both the database are same or not.
 900     if(tdbFields!=csqlFields){
 901         printError(ErrSysInit,"Number of fields between CSQL and TDB are not equal.");
 902         SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 903         return ErrSysInit;
 904     }
 905     retValue=SQLAllocHandle (SQL_HANDLE_STMT, hdbc, &hstmtmeta);
 906     if(retValue){
 907         printError(ErrSysInit, "Unable to allocate ODBC handle. \n");
 908         SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 909         return ErrSysInit;
 910     }
 911     retValue=SQLPrimaryKeys(hstmtmeta, NULL, SQL_NTS, NULL, SQL_NTS, (SQLCHAR*) tableName, SQL_NTS);
 912     retValue = SQLBindCol(hstmtmeta, 4, SQL_C_CHAR,columnname, 129,NULL);
 913
 914     while(icol<=tdbFields){
 915         retValue = SQLDescribeCol(hstmt, icol, colName, colNameMax,
 916                              &nameLength, &colType, &colLength,
 917                              &scale, &nullable);//TDB Field Name
 918         if(retValue){
 919             printError(ErrSysInit, "Unable to retrieve ODBC column info.\n");
 920             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 921             return ErrSysInit;
 922         }
 923         Util::str_tolower((char*)colName);
 924         elem = (Identifier*) fNameIter.nextElement();
 925         sqlStmt->getFieldInfo(tableName, (const char*)elem->name, info);
 926         char fldName[20];
 927         int isNull;
 928         int isPrimary;
 929         rv = stmt->getParamFldInfo(icol,info);
 930         char *name=(info->fldName);//Getting field name for CSQL table.
 931         Util::str_tolower((char*)name);
 932         if(strcmp(name,(char *)colName) != 0){ //Field name matching between CSQL and TDB.
 933             printError(ErrSysInit,"CSQL's '%s' field did not match with TDB's '%s' field.\n",name,(char*)colName);
 934             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 935             return ErrSysInit;
 936         }
 937
 938         // DataType matching between CSQL and TDB
 939         char ptr[IDENTIFIER_LENGTH]; ptr[0]='\0';
 940         char ptr1[IDENTIFIER_LENGTH]; ptr1[0]='\0';
 941
 942         sprintf(ptr,"%s",AllDataType::getSQLString (AllDataType::convertFromSQLType(colType,colLength,scale,tdbName)));
 943         sprintf(ptr1,"%s",AllDataType::getSQLString(info->type));//CSQL Type
 944         if(strcmp(ptr,ptr1)!=0){
 945            printError(ErrSysInit,"DataType did not match for '%s' field in CSQL.\n",name);
 946            SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 947            return ErrSysInit;
 948         }
 949
 950        // Primary Key checking
 951        bool tdbPKey=false;
 952        if(SQLFetch( hstmtmeta ) == SQL_SUCCESS) tdbPKey=true;
 953        if(tdbPKey && (!info->isPrimary))
 954        printf("Warning: In CSQL, The %s's '%s' field should have Primery Key constraint.\n",tableName,name);
 955        if((!tdbPKey) && info->isPrimary)
 956            printf("Warning: In TDB, The %s's '%s' field should have Primary Key constraint.\n",tableName,colName);
 957
 958        // NotNull Checking
 959        bool isCsqlNotNull=false;
 960        bool isTdbNotNull=false;
 961        if(tdbName==mysql){
 962            if(info->isNull && nullable)
 963            printf("Warning: In TDB, The %s's '%s' field should have a NOT NULL constraint.\n",tableName,colName);
 964            if((!info->isNull) && (!nullable))
 965                printf("Warning: In CSQL, The %s's '%s' field should have a NOT NULL constraint.\n",tableName,name);
 966        }
 967        icol++;
 968    }
 969    return OK;
 970 }
 971
 972 DbRetVal CacheTableLoader::cacheAllTablesFromDs(char *dsnName,bool tableDefinition, bool isDirect,char *username, char *password)
 973 {
 974     char dsn[72];
 975     DbRetVal rv = OK;
 976     FILE *fp;
 977     fp = fopen(Conf :: config.getDsConfigFile(),"r");
 978     if(fp==NULL) {
 979         printError(ErrSysInit, "csqlds.conf file does not exist");
 980         return ErrSysInit;
 981     }
 982     char dsnId[IDENTIFIER_LENGTH]; dsnId[0]='\0';
 983     char user[IDENTIFIER_LENGTH]; user[0] = '\0';
 984     char passwd[IDENTIFIER_LENGTH]; passwd[0] = '\0';
 985     char tdb[IDENTIFIER_LENGTH]; tdb[0]='\0';
 986     unsigned int mode;
 987     bool isCached=false;
 988
 989     // If -d option is disable, the If statementn will true.
 990     if(strcmp(dsnName,"")==0) {
 991         strcpy(dsnName, Conf::config.getDSN());
 992     }
 993     bool isDSNExist=false;
 994     while(!feof(fp)) {
 995        fscanf(fp,"%s %s %s %s\n",dsnId,user,passwd,tdb);
 996        if(strcmp(dsnId,dsnName)==0) {
 997            if( strcmp(user,"NULL")!=0 && strcmp(passwd,"NULL")!=0) {
 998                sprintf(dsn,"DSN=%s;UID=%s;PWD=%s;",dsnName,user,passwd);
 999                isDSNExist=true;
1000                break;
1001            }else{
1002                sprintf(dsn,"DSN=%s;",dsnName);
1003                isDSNExist=true;
1004                break;
1005            }
1006        }
1007     }
1008      if(!isDSNExist) {
1009        printError(ErrNotExists,"Entries is not present in the csqlds.conf file\n");
1010        fclose(fp);
1011        return ErrNotExists;
1012     }
1013     fclose(fp);
1014
1015     TDBInfo tdbName=mysql;
1016     if (strcasecmp(tdb,"mysql") == 0) tdbName=mysql;
1017     else if (strcasecmp(tdb,"postgres")==0) tdbName=postgres;
1018     else printError(ErrNotFound,"Target Database Name is not properly set.Tdb name could be MySql and Postgres.\n");
1019     logFine(Conf::logger, "TDB Name:%s\n", tdb);
1020
1021     // The ODBC section in intended to get all the tables from TDB,
1022     // what SQLTables() is doing that.
1023
1024     SQLCHAR outstr[1024];
1025     SQLSMALLINT outstrlen;
1026     int retValue =0;
1027     SQLHENV henv;
1028     SQLHDBC hdbc;
1029     SQLHSTMT hstmt;
1030     SQLSMALLINT columns;
1031     char table[IDENTIFIER_LENGTH][IDENTIFIER_LENGTH];
1032     int counter=0;
1033     char buf[IDENTIFIER_LENGTH];
1034     int row = 0;
1035     SQLINTEGER indicator[ 5 ];
1036      int colPos;//Only to bind table name filed.
1037
1038     CacheTableLoader cacheLoader;
1039
1040     retValue = SQLAllocHandle (SQL_HANDLE_ENV, SQL_NULL_HANDLE, &henv);
1041     if (retValue) {
1042         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
1043         return ErrSysInit;
1044     }
1045     // We want ODBC 3 support
1046     SQLSetEnvAttr(henv, SQL_ATTR_ODBC_VERSION, (void *) SQL_OV_ODBC3, 0);
1047     retValue = SQLAllocHandle (SQL_HANDLE_DBC, henv, &hdbc);
1048      if (retValue) {
1049         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
1050         return ErrSysInit;
1051     }
1052     retValue = SQLDriverConnect(hdbc, NULL, (SQLCHAR*)dsn, SQL_NTS,
1053                          outstr, sizeof(outstr), &outstrlen,
1054                          SQL_DRIVER_NOPROMPT);
1055     if (SQL_SUCCEEDED(retValue)) {
1056         printDebug(DM_Gateway, "Connected to target database using dsn = %s\n", dsn);
1057     }else{
1058         printError(ErrSysInit, "Failed to connect to target database\n");
1059         return ErrSysInit;
1060     }
1061     retValue=SQLAllocHandle (SQL_HANDLE_STMT, hdbc, &hstmt);
1062     if (retValue) {
1063         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
1064         return ErrSysInit;
1065     }
1066     if(tdbName == mysql){
1067         colPos=3;
1068         // User name is required in upper case for the SQLTables()'s  4th parameter
1069         Util::str_toupper((char*)user);
1070         retValue=SQLTables(hstmt,NULL, 0, (SQLCHAR*)user, SQL_NTS, NULL, 0, (SQLCHAR*)"TABLE", SQL_NTS);
1071         if(retValue){
1072             printError(ErrSysInit, "Unable to retrieve list of tables\n");
1073             return ErrSysInit;
1074         }
1075         // Binding Column for 3rd parameter to get Table name.
1076         retValue=SQLBindCol(hstmt,3, SQL_C_CHAR,buf,sizeof(buf),NULL);
1077         if(retValue){
1078             printError(ErrSysInit,"Unable to BindCol\n");
1079             return ErrSysInit;
1080         }
1081         // For Postgres DB , SQLTables() retrieves all system and metadata tables,along with User defined table.
1082         // So Here is a another option to fetch the user defined tables only
1083     }else if(tdbName==postgres){
1084         SQLCHAR table[200]="SELECT table_name FROM information_schema.tables  WHERE table_schema NOT IN ('pg_catalog','information_schema');";
1085         retValue=SQLPrepare(hstmt,table,SQL_NTS);
1086         if(retValue){
1087             printError(ErrSysInit,"Unable to Prapare the statement\n");
1088             return ErrSysInit;
1089         }
1090         retValue = SQLBindCol(hstmt,1,SQL_C_CHAR,buf,sizeof(buf),NULL);
1091         if(retValue){
1092             printError(ErrSysInit,"Unable to bind the column\n");
1093             return ErrSysInit;
1094         }
1095         retValue = SQLExecute(hstmt);
1096         if(retValue){
1097             printError(ErrSysInit,"Unable to execute\n");
1098             return ErrSysInit;
1099         }
1100     }
1101
1102     while(SQL_SUCCEEDED(retValue = SQLFetch(hstmt))){
1103         // copy Buffer value
1104         //strcpy(&table[counter][0],buf);
1105         cacheLoader.setDsnName(dsnName);
1106         TableConf::config.setDsnName(dsnName);
1107         cacheLoader.setConnParam(username, password);
1108         TableConf::config.setConnParam(username, password);
1109         // Check table is cached or not
1110         mode = TableConf::config.getTableMode(buf);
1111         cacheLoader.setTable(buf);
1112         TableConf::config.setTable(buf);
1113         isCached = TableConf::config.isTableCached(mode);
1114         if(isCached){
1115             printf("Warning: Table '%s' is already cached.\n",buf);
1116         }else{
1117              rv = cacheLoader.load(tableDefinition);
1118              if(rv != OK){
1119                  printf("Warning: Table '%s' is present in CSQL locally.\n",buf);
1120              }else{
1121                   TableConf::config.addToCacheTableFile(isDirect);
1122                   printf("Cached Table:%s\n",buf);
1123                   TableConf::config.init();
1124              }
1125         }
1126         counter++;
1127     }
1128     // Checking couter value
1129     if(counter==0)
1130     printf("There is no table present in Target Database.\n");
1131     retValue=SQLCloseCursor(hstmt);
1132     if(retValue){
1133         printError(ErrSysInit,"Unable to close the cursor\n");
1134         return ErrSysInit;
1135     }
1136     retValue=SQLTransact(henv,hdbc,SQL_COMMIT);
1137     if(retValue){
1138         printError(ErrSysInit,"Unable to commit the transaction\n");
1139         return ErrSysInit;
1140     }
1141     retValue = SQLFreeHandle(SQL_HANDLE_STMT,hstmt);
1142     if(retValue){
1143         printError(ErrSysInit,"Unable to free statement handle\n");
1144         return ErrSysInit;
1145     }
1146     retValue = SQLDisconnect(hdbc);
1147     if(retValue){
1148          printError(ErrSysInit,"Unable to disconnect from DS handle\n");
1149          return ErrSysInit;
1150     }
1151     retValue = SQLFreeHandle(SQL_HANDLE_DBC,hdbc);
1152     if(retValue){
1153         printError(ErrSysInit,"Unable to free connection handle\n");
1154         return ErrSysInit;
1155     }
1156     retValue = SQLFreeHandle(SQL_HANDLE_ENV,henv);
1157     if(retValue){
1158         printError(ErrSysInit,"Unable to free environment handle\n");
1159         return ErrSysInit;
1160     }
1161     return OK;
1162 }
1163
1164 bool CacheTableLoader::resolveForDSN(char *dsn, TDBInfo &tdbName, DbRetVal &rv)
1165 {
1166     char dsnId[IDENTIFIER_LENGTH]; dsnId[0]='\0';
1167     char user[IDENTIFIER_LENGTH]; user[0] = '\0';
1168     char passwd[IDENTIFIER_LENGTH]; passwd[0] = '\0';
1169     char tdb[IDENTIFIER_LENGTH]; tdb[0]='\0';
1170     FILE *fp;
1171
1172     fp = fopen(Conf::config.getDsConfigFile(),"r");
1173     if(fp==NULL) {
1174         printError(ErrSysInit, "csqlds.conf file does not exist");
1175         rv = ErrSysInit;
1176         return false;
1177     }
1178
1179     // STARTs Here:
1180     // DSN, user and password value is read here from csql.conf file and
1181     // csqlds.conf file.
1182     // it's true if -d option is specified and the DSN value not matched with
1183     // csql.conf's DSN.
1184     if(strcmp(dsnName,"")==0) strcpy(dsnName, Conf::config.getDSN());
1185
1186     bool isDSNExist=false;
1187     while(!feof(fp)) {
1188         fscanf(fp,"%s %s %s %s\n",dsnId,user,passwd,tdb);
1189     if(strcmp(dsnId,dsnName)==0) {   // Both the DSN is matched here
1190         if(strcmp(user,"NULL")!=0 && strcmp(passwd,"NULL")!=0) {
1191                 sprintf(dsn,"DSN=%s;UID=%s;PWD=%s;",dsnName,user,passwd);
1192                 isDSNExist=true;
1193                 break;
1194             } else {
1195                 sprintf(dsn,"DSN=%s;",dsnName);
1196                 isDSNExist=true;
1197                 break;
1198             }
1199         }
1200     }
1201     if(!isDSNExist) {
1202         printError(ErrNotExists,"Entries is not present in the csqlds.conf file\n");
1203         fclose(fp);
1204         rv = ErrNotExists;
1205         return false;
1206     }
1207     fclose(fp);
1208     if (strcasecmp(tdb,"mysql") == 0) tdbName=mysql;
1209     else if (strcasecmp(tdb,"postgres")==0) tdbName=postgres;
1210     else {
1211         printError(ErrNotFound," Target Database Name is not properly set.Tdb name could be mysql, postgres\n");
1212         rv = ErrNotFound;
1213         return false;
1214     }
1215     logFine(Conf::logger, "TDB Name:%s\n", tdb);
1216     return isDSNExist;
1217 }
1218
1219 void CacheTableLoader::generateCacheTableStatement(char *stmtBuf)
1220 {
1221     if(((strcmp(conditionVal,"")==0) || (strcmp(conditionVal,"NULL")==0)) &&
1222        ((strcmp(fieldlistVal,"")==0) || (strcmp(fieldlistVal,"NULL")==0)))
1223     {
1224        sprintf(stmtBuf, "SELECT * FROM %s;", tableName);
1225     }
1226     else if(((strcmp(conditionVal,"")!=0) || (strcmp(conditionVal,"NULL")!=0)) &&
1227             ((strcmp(fieldlistVal,"")==0) || (strcmp(fieldlistVal,"NULL")==0)))
1228     {
1229        sprintf(stmtBuf,"SELECT * FROM %s where %s;",tableName,conditionVal);
1230
1231     }
1232     else if(((strcmp(conditionVal,"")==0) || (strcmp(conditionVal,"NULL")==0)) &&
1233             ((strcmp(fieldlistVal,"")!=0) || (strcmp(fieldlistVal,"NULL")!=0)))
1234     {
1235         sprintf(stmtBuf,"SELECT %s FROM %s;",fieldlistVal,tableName);
1236     }
1237     else {
1238         sprintf(stmtBuf,"SELECT %s FROM %s where %s;",fieldlistVal,tableName,conditionVal);
1239     }
1240     logFinest(Conf::logger, "Cache Table Stmt %s", stmtBuf);
1241 }
1242
1243 bool CacheTableLoader::prepareCreateIndexStatement(SQLHSTMT hstmtmeta,
1244                                     char *crtIdxStmt, TDBInfo tdbName, HashIndexInitInfo *inf)
1245 {
1246     char columnname[IDENTIFIER_LENGTH];
1247     char indexname[IDENTIFIER_LENGTH];
1248     int retValue=SQLPrimaryKeys(hstmtmeta, NULL, SQL_NTS, NULL, SQL_NTS,
1249                                                (SQLCHAR*) tableName, SQL_NTS);
1250     retValue = SQLBindCol(hstmtmeta, 4, SQL_C_CHAR,columnname, 129,NULL);
1251     char *ptr=crtIdxStmt;
1252     sprintf(ptr, "CREATE INDEX %s_PRIMARY on %s ( ", tableName, tableName);
1253     ptr += strlen(ptr);
1254     bool isPriIndex=false;
1255     char indname[IDENTIFIER_LENGTH];
1256     if(SQLFetch( hstmtmeta ) == SQL_SUCCESS)
1257     {
1258         Util::str_tolower(columnname);
1259         inf->list.append(columnname);
1260         sprintf(ptr, "%s ", columnname);
1261         ptr += strlen(ptr);
1262         while ( SQLFetch( hstmtmeta ) == SQL_SUCCESS ) {
1263             Util::str_tolower(columnname);
1264             inf->list.append(columnname);
1265             sprintf(ptr, ", %s ", columnname);
1266             ptr += strlen(ptr);
1267         }
1268         sprintf(ptr, ") PRIMARY SIZE 10007;");
1269         inf->indType =  hashIndex;
1270         inf->bucketSize = 10007;
1271         inf->isUnique = true; inf->isPrimary = true;
1272         strcpy(inf->tableName, tableName);
1273         strcpy(indexname,"PRIMARY");
1274         sprintf(indname, "%s_%s", tableName, indexname);
1275         isPriIndex=true;
1276     }
1277     return isPriIndex;
1278 }
1279
1280 DbRetVal CacheTableLoader::prepareCreateTableStatement(char *crtTblStmt, SQLHSTMT hstmt,
1281                   HashIndexInitInfo *inf, int totalFields, TDBInfo tdbName, bool &isKeyFld)
1282 {
1283     DbRetVal rv = OK;
1284     UWORD icol=1;
1285     UCHAR colName[IDENTIFIER_LENGTH];
1286     SWORD colNameMax=IDENTIFIER_LENGTH;
1287     SWORD nameLength=0;
1288     SWORD colType=0;
1289     SQLULEN colLength = 0;
1290     SWORD scale=0;
1291     SWORD nullable=0;
1292     int retValue = 0;
1293     bool isNullfld=false;
1294     bool firstFld = true;
1295     char *name = NULL;
1296
1297     char *ptr = crtTblStmt;
1298     sprintf(ptr, "CREATE TABLE %s ( ", tableName);
1299     ptr += strlen(ptr);
1300     while (icol <= totalFields) {
1301         retValue = SQLDescribeCol(hstmt, icol, colName, colNameMax,
1302                                          &nameLength, &colType, &colLength,
1303                                                             &scale, &nullable);
1304         if (retValue) {
1305             printError(ErrSysInit, "Unable to retrieve ODBC column info\n");
1306             return ErrSysInit;
1307         }
1308         Util::str_tolower((char*)colName);
1309         printDebug(DM_Gateway, "Describe Column %s %d %d %d %d\n",
1310                       colName, colType, colLength, scale, nullable);
1311         logFinest(Conf::logger, "Describe Column colName:%s \
1312                                 colType:%d colLen:%d scale:%d nullable:%d\n",
1313                                 colName, colType, colLength, scale, nullable);
1314         icol++;
1315         if(strcmp((char*)colName,fieldName)== 0)
1316         {
1317             isKeyFld=true;
1318             isNullfld=true;
1319         }
1320         bool isPriFld=false;
1321         if (nullable) {
1322             inf->list.resetIter();
1323             while ((name=inf->list.nextFieldName())!=NULL) {
1324                 if(0==strcmp((char*)colName,name)) {
1325                     if (firstFld) {
1326                         firstFld = false;
1327                         sprintf(ptr, "%s %s", colName, AllDataType::getSQLString(
1328                                                        AllDataType::convertFromSQLType(
1329                                                        colType,colLength,scale,tdbName)));
1330                         ptr += strlen(ptr);
1331                         if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY)
1332                         {
1333                             sprintf(ptr, "(%d) NOT NULL",colLength+1);
1334                         } else { sprintf(ptr, " NOT NULL"); }
1335                             ptr += strlen(ptr);
1336                     } else {
1337                         sprintf(ptr, ", %s %s", colName, AllDataType::getSQLString(
1338                                                          AllDataType::convertFromSQLType(
1339                                                        colType,colLength,scale,tdbName)));
1340                         ptr += strlen(ptr);
1341                         if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY)
1342                         {
1343                             sprintf(ptr, "(%d) NOT NULL",colLength+1);
1344                         } else { sprintf(ptr, " NOT NULL"); }
1345                         ptr += strlen(ptr);
1346                     }
1347                     //tabDef.addField((char*)colName, AllDataType::convertFromSQLType(
1348                     //                                  colType,colLength,scale,tdbName),
1349                     //                                  colLength +1, NULL, true);
1350                     isPriFld=true;
1351                     break;
1352                 }
1353             }
1354             if(!isPriFld) {
1355                 if(!isNullfld) {
1356                     if (firstFld) {
1357                         firstFld = false;
1358                         sprintf(ptr, "%s %s", colName, AllDataType::getSQLString(
1359                                                        AllDataType::convertFromSQLType(
1360                                                        colType,colLength,scale,tdbName)));
1361                         ptr += strlen(ptr);
1362                         if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
1363                             sprintf(ptr, "(%d)",colLength+1);
1364                             ptr += strlen(ptr);
1365                         }
1366                     } else {
1367                         sprintf(ptr, ", %s %s", colName, AllDataType::getSQLString(
1368                                                          AllDataType::convertFromSQLType(
1369                                                          colType,colLength,scale,tdbName)));
1370                         ptr += strlen(ptr);
1371                         if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
1372                             sprintf(ptr, "(%d)",colLength+1);
1373                             ptr += strlen(ptr);
1374                         }
1375                     }
1376                     //tabDef.addField((char*)colName, AllDataType::convertFromSQLType(
1377                     //                                colType,colLength,scale,tdbName), colLength+1);
1378                 } else {
1379                     if (firstFld) {
1380                         firstFld = false;
1381                         sprintf(ptr, "%s %s", colName, AllDataType::getSQLString(
1382                                                        AllDataType::convertFromSQLType(
1383                                                        colType,colLength,scale,tdbName)));
1384                         ptr += strlen(ptr);
1385                         if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
1386                             sprintf(ptr, "(%d) NOT NULL",colLength+1);
1387                         } else { sprintf(ptr, " NOT NULL"); }
1388                         ptr += strlen(ptr);
1389                     } else {
1390                         sprintf(ptr, ", %s %s", colName, AllDataType::getSQLString(
1391                                                          AllDataType::convertFromSQLType(
1392                                                          colType,colLength,scale,tdbName)));
1393                         ptr += strlen(ptr);
1394                         if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
1395                             sprintf(ptr, "(%d) NOT NULL",colLength+1);
1396                         } else { sprintf(ptr, " NOT NULL"); }
1397                         ptr += strlen(ptr);
1398                     }
1399                     //tabDef.addField((char*)colName, AllDataType::convertFromSQLType(
1400                     //                                  colType,colLength,scale,tdbName),
1401                     //                                  colLength+1, NULL, true);
1402                     isNullfld=false;
1403                 }
1404             }
1405         } else {
1406             if (firstFld) {
1407                 firstFld = false;
1408                 sprintf(ptr, "%s %s", colName, AllDataType::getSQLString(
1409                                                AllDataType::convertFromSQLType(
1410                                                colType,colLength,scale,tdbName)));
1411                 ptr += strlen(ptr);
1412                 if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
1413                     sprintf(ptr, "(%d) NOT NULL",colLength+1);
1414                 } else { sprintf(ptr, " NOT NULL"); }
1415                 ptr += strlen(ptr);
1416             } else {
1417                 sprintf(ptr, ", %s %s", colName, AllDataType::getSQLString(
1418                                                  AllDataType::convertFromSQLType(
1419                                                  colType,colLength, scale, tdbName)));
1420                 ptr += strlen(ptr);
1421                 if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
1422                     sprintf(ptr, "(%d) NOT NULL",colLength+1);
1423                 } else { sprintf(ptr, " NOT NULL"); }
1424                 ptr += strlen(ptr);
1425             }
1426             //tabDef.addField((char*)colName, AllDataType::convertFromSQLType(
1427             //                                  colType,colLength,scale, tdbName),
1428             //                                  colLength +1, NULL, true);
1429         }
1430     }
1431     sprintf(ptr, ");");
1432     ptr += strlen(ptr);
1433     //printf("table stmt '%s'\n", crtTblStmt);
1434     return rv;
1435 }
1436
1437 void CacheTableLoader::prepareInsertStatement(AbsSqlStatement *stmt, List *fNameList, char *insStmt)
1438 {
1439     DbRetVal rv = OK;
1440     char *ptr = insStmt;
1441     sprintf(ptr,"INSERT INTO %s VALUES(", tableName);
1442     ptr += strlen(ptr);
1443     bool firstFld = true;
1444     SqlStatement *sqlStmt = (SqlStatement *)stmt;
1445     *fNameList = sqlStmt->getFieldNameList(tableName, rv);
1446     int noOfFields = fNameList->size();
1447     while (noOfFields--) {
1448         if (firstFld) {
1449             firstFld = false;
1450             sprintf(ptr,"?", tableName);
1451             ptr += strlen(ptr);
1452         } else {
1453             sprintf(ptr, ",?");
1454             ptr += strlen(ptr);
1455         }
1456     }
1457     sprintf(ptr, ");");
1458     ptr += strlen(ptr);
1459 }