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
  26
  27 DbRetVal CacheTableLoader::checkSecondTimeSqlPrimaryKeys(SQLHSTMT hstmtmeta,char *tableName,char *ptr,HashIndexInitInfo *inf,bool &isPriIndex)
  28 {
  29         int retValue =0;
  30         char columnname[IDENTIFIER_LENGTH];
  31         retValue=SQLPrimaryKeys(hstmtmeta, NULL, SQL_NTS, NULL, SQL_NTS, (SQLCHAR*) tableName, SQL_NTS);
  32         retValue = SQLBindCol(hstmtmeta, 4, SQL_C_CHAR,columnname, 129,NULL);
  33         char indname[IDENTIFIER_LENGTH];
  34         if(SQLFetch( hstmtmeta ) == SQL_SUCCESS)
  35         {
  36             Util::str_tolower(columnname);
  37             inf->list.append(columnname);
  38             sprintf(ptr, "%s ", columnname);
  39             ptr += strlen(ptr);
  40             while ( SQLFetch( hstmtmeta ) == SQL_SUCCESS ) {
  41                 Util::str_tolower(columnname);
  42                 inf->list.append(columnname);
  43                 sprintf(ptr, ", %s ", columnname);
  44                 ptr += strlen(ptr);
  45             }
  46             sprintf(ptr, ") PRIMARY SIZE 10007;");
  47             isPriIndex=true;
  48         }
  49         return OK;
  50 }
  51
  52 DbRetVal CacheTableLoader::load(bool tabDefinition)
  53 {
  54     AbsSqlConnection *conn = SqlFactory::createConnection(CSqlLog);
  55     DbRetVal rv = conn->connect(userName, password);
  56     if (rv != OK) { delete conn; return ErrSysInit; }
  57     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlLog);
  58     stmt->setConnection(conn);
  59     SqlLogConnection *logConn = (SqlLogConnection *) conn;
  60     logConn->setNoMsgLog(true);
  61     SqlConnection *con = (SqlConnection *) conn->getInnerConnection();
  62     DatabaseManager *dbMgr = con->getConnObject().getDatabaseManager();
  63     if (tabDefinition == false) {
  64         Table *tbl = dbMgr->openTable(tableName);
  65         if (tbl == NULL) {
  66             conn->disconnect();
  67             delete stmt;
  68             delete conn;
  69             return ErrNotExists;
  70         }
  71         if (tbl->numTuples()) {
  72             printError(ErrNotEmpty, "The table '\%s\' is not empty", tableName);
  73             dbMgr->closeTable(tbl);
  74             conn->disconnect();
  75             delete stmt;
  76             delete conn;
  77             return ErrNotEmpty;
  78         }
  79         dbMgr->closeTable(tbl);
  80     }
  81     conn->beginTrans();
  82     rv = load(conn, stmt, tabDefinition);
  83     conn->commit();
  84     stmt->free();
  85     conn->disconnect();
  86     delete stmt;
  87     delete conn;
  88     return rv;
  89 }
  90
  91 DbRetVal CacheTableLoader::load(AbsSqlConnection *conn, AbsSqlStatement *stmt, bool tabDefinition)
  92 {
  93     char dsn[72];
  94     DbRetVal rv = OK;
  95     FILE *fp;
  96     fp = fopen(Conf :: config.getDsConfigFile(),"r");
  97     if(fp==NULL) {
  98         printError(ErrSysInit, "csqlds.conf file does not exist");
  99             return ErrSysInit;
 100     }
 101     char dsnId[IDENTIFIER_LENGTH]; dsnId[0]='\0';
 102     char user[IDENTIFIER_LENGTH]; user[0] = '\0';
 103     char passwd[IDENTIFIER_LENGTH]; passwd[0] = '\0';
 104     char tdb[IDENTIFIER_LENGTH]; tdb[0]='\0';
 105
 106     // STARTs Here:
 107     // DSN, user and password value is read here from csql.conf fiel and csqlds.conf file.
 108
 109     if(strcmp(dsnName,"")==0) { // it's true if -d option is specified and the DSN value not matched with csql.conf's DSN.
 110         strcpy(dsnName, Conf::config.getDSN());
 111     }
 112     bool isDSNExist=false;
 113     while(!feof(fp)) {
 114         fscanf(fp,"%s %s %s %s\n",dsnId,user,passwd,tdb);
 115             if(strcmp(dsnId,dsnName)==0) {   // Both the DSN is matched here
 116                 if( strcmp(user,"NULL")!=0 && strcmp(passwd,"NULL")!=0) {
 117                     sprintf(dsn,"DSN=%s;UID=%s;PWD=%s;",dsnName,user,passwd);
 118                 isDSNExist=true;
 119                 break;
 120                 } else {
 121                     sprintf(dsn,"DSN=%s;",dsnName);
 122                 isDSNExist=true;
 123                 break;
 124                 }
 125         }
 126     }
 127     if(!isDSNExist) {
 128         printError(ErrNotExists,"Entries is not present in the csqlds.conf file\n");
 129         fclose(fp);
 130         return ErrNotExists;
 131     }
 132     fclose(fp);
 133     TDBInfo tdbName=mysql;
 134     if (strcmp(tdb,"postgres")==0) tdbName=postgres;
 135     else tdbName=mysql;
 136
 137   //ENDs Here:
 138
 139
 140     //DatabaseManager *dbMgr = (DatabaseManager *) conn->getDatabaseManager();
 141     //char dsn[72];
 142     SqlConnection *con = (SqlConnection *) conn->getInnerConnection();
 143     DatabaseManager *dbMgr = con->getConnObject().getDatabaseManager();
 144
 145     SQLCHAR outstr[1024];
 146     SQLSMALLINT outstrlen;
 147     int retValue =0;
 148     SQLHENV henv;
 149     SQLHDBC hdbc;
 150     SQLHSTMT hstmt;
 151     retValue = SQLAllocHandle (SQL_HANDLE_ENV, SQL_NULL_HANDLE, &henv);
 152     if (retValue) {
 153         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 154         return ErrSysInit;
 155     }
 156     SQLSetEnvAttr(henv, SQL_ATTR_ODBC_VERSION, (void *) SQL_OV_ODBC3, 0);
 157     retValue = SQLAllocHandle (SQL_HANDLE_DBC, henv, &hdbc);
 158     if (retValue) {
 159         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 160         return ErrSysInit;
 161     }
 162     retValue = SQLDriverConnect(hdbc, NULL, (SQLCHAR*)dsn, SQL_NTS,
 163                          outstr, sizeof(outstr), &outstrlen,
 164                          SQL_DRIVER_NOPROMPT);
 165     if (SQL_SUCCEEDED(retValue)) {
 166         printDebug(DM_Gateway, "Connected to target database using dsn = %s\n", dsn);
 167     } else {
 168         printError(ErrSysInit, "Failed to connect to target database\n");
 169         return ErrSysInit;
 170     }
 171
 172     retValue=SQLAllocHandle (SQL_HANDLE_STMT, hdbc, &hstmt);
 173     if (retValue) {
 174         printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 175         return ErrSysInit;
 176     }
 177     char stmtBuf[1024];
 178
 179     if(((strcmp(conditionVal,"")==0) || (strcmp(conditionVal,"NULL")==0)) && ((strcmp(fieldlistVal,"")==0) || (strcmp(fieldlistVal,"NULL")==0)))
 180     {
 181        sprintf(stmtBuf, "SELECT * FROM %s;", tableName);
 182     }
 183     else if(((strcmp(conditionVal,"")!=0) || (strcmp(conditionVal,"NULL")!=0)) && ((strcmp(fieldlistVal,"")==0) || (strcmp(fieldlistVal,"NULL")==0)))
 184     {
 185        sprintf(stmtBuf,"SELECT * FROM %s where %s;",tableName,conditionVal);
 186
 187     }
 188     else if(((strcmp(conditionVal,"")==0) || (strcmp(conditionVal,"NULL")==0)) && ((strcmp(fieldlistVal,"")!=0) || (strcmp(fieldlistVal,"NULL")!=0)))
 189     {
 190         sprintf(stmtBuf,"SELECT %s FROM %s;",fieldlistVal,tableName);
 191     }
 192     else
 193         sprintf(stmtBuf,"SELECT %s FROM %s where %s;",fieldlistVal,tableName,conditionVal);
 194
 195     retValue = SQLPrepare (hstmt, (unsigned char *) stmtBuf, SQL_NTS);
 196     if (retValue) {
 197         printError(ErrSysInit, "Unable to Prepare ODBC statement \n");
 198         return ErrSysInit;
 199     }
 200     if (tabDefinition) {
 201         short totalFields=0;
 202         retValue = SQLNumResultCols (hstmt, &totalFields);
 203         if (retValue) {
 204             printError(ErrSysInit, "Unable to retrieve ODBC total columns\n");
 205             return ErrSysInit;
 206         }
 207         UWORD                   icol;
 208         UCHAR                   colName[IDENTIFIER_LENGTH];
 209         SWORD                   colNameMax;
 210         SWORD                   nameLength;
 211         SWORD                   colType;
 212         SQLULEN                 colLength = 0;
 213         SWORD                   scale;
 214         SWORD                   nullable;
 215         TableDef tabDef;
 216         icol = 1;
 217         colNameMax = IDENTIFIER_LENGTH;
 218         char columnname[IDENTIFIER_LENGTH];
 219         char indexname[IDENTIFIER_LENGTH];
 220         short type; short unique;
 221         SQLHSTMT hstmtmeta;
 222         retValue=SQLAllocHandle (SQL_HANDLE_STMT, hdbc, &hstmtmeta);
 223         if (retValue)
 224         {
 225             printError(ErrSysInit, "Unable to allocate ODBC handle \n");
 226             return ErrSysInit;
 227         }
 228
 229         retValue=SQLPrimaryKeys(hstmtmeta, NULL, SQL_NTS, NULL, SQL_NTS, (SQLCHAR*) tableName, SQL_NTS);
 230         retValue = SQLBindCol(hstmtmeta, 4, SQL_C_CHAR,columnname, 129,NULL);
 231         HashIndexInitInfo *inf = new HashIndexInitInfo();
 232         char crtIdxStmt[1024];
 233         char *name = NULL;
 234         char *ptr=crtIdxStmt;
 235         sprintf(ptr, "CREATE INDEX %s_PRIMARY on %s ( ", tableName, tableName);
 236         ptr += strlen(ptr);
 237         bool isPriIndex=false;
 238         char indname[IDENTIFIER_LENGTH];
 239         if(SQLFetch( hstmtmeta ) == SQL_SUCCESS)
 240         {
 241             Util::str_tolower(columnname);
 242             inf->list.append(columnname);
 243             sprintf(ptr, "%s ", columnname);
 244             ptr += strlen(ptr);
 245             while ( SQLFetch( hstmtmeta ) == SQL_SUCCESS ) {
 246                 Util::str_tolower(columnname);
 247                 inf->list.append(columnname);
 248                 sprintf(ptr, ", %s ", columnname);
 249                 ptr += strlen(ptr);
 250             }
 251             sprintf(ptr, ") PRIMARY SIZE 10007;");
 252             inf->indType =  hashIndex;
 253             inf->bucketSize = 10007;
 254             inf->isUnique = true; inf->isPrimary = true;
 255             strcpy(inf->tableName, tableName);
 256             strcpy(indexname,"PRIMARY");
 257             sprintf(indname, "%s_%s", tableName, indexname);
 258             isPriIndex=true;
 259         }
 260         bool iskeyfieldExist=false;
 261         bool isPKFieldSpecified = false;
 262         if((strcmp(fieldName,"")!=0) && (strcmp(fieldName,"NULL")!=0) )
 263         {
 264             isPKFieldSpecified = true;
 265         }
 266         if ( isPriIndex && ( strcmp(fieldlistVal,"")!=0 ) &&
 267                                        ( strcmp(fieldlistVal,"NULL") != 0 )) {
 268             inf->list.resetIter();
 269             while ( (name=inf->list.nextFieldName()) != NULL) {
 270                 iskeyfieldExist = TableConf::config.isFieldExist(name);
 271                 if(!iskeyfieldExist) { break; }
 272             }
 273         } else if (isPriIndex) { iskeyfieldExist = true; }
 274         if ( isPKFieldSpecified &&  !(TableConf::config.isFieldExist(fieldName)) )
 275         {
 276             if ( Conf::config.useTwoWayCache() &&
 277                              (strcmp(fieldlistVal,"")!=0) &&
 278                                          (strcmp(fieldlistVal,"NULL")!=0))
 279             {
 280                 printError(ErrSysInit, "Bidirectonal caching should have primary key in %s \n", tableName);
 281                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 282                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 283                 SQLDisconnect (hdbc);
 284                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 285                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 286                 delete inf;
 287                 return ErrSysInit;
 288             }
 289         }
 290         if (!iskeyfieldExist && !isPKFieldSpecified )
 291         {
 292             if(Conf::config.useTwoWayCache())
 293             {
 294                 printError(ErrSysInit, "Bidirectonal caching fail for no primary key in %s \n", tableName);
 295                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 296                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 297                 SQLDisconnect (hdbc);
 298                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 299                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 300                 delete inf;
 301                 return ErrSysInit;
 302             }
 303         }
 304
 305 /*        if(isPriIndex) ;
 306         else if (Conf::config.useTwoWayCache() && !iskeyfieldExist) {
 307             printError(ErrSysInit, "Bidirectonal caching fail for no primary key in %s \n", tableName);
 308             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 309             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 310             SQLDisconnect (hdbc);
 311             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 312             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 313             return ErrSysInit;
 314         }
 315 */
 316         bool isKeyFld=false;
 317         bool isNullfld=false;
 318         bool firstFld = true;
 319         char crtTblStmt[1024];
 320         ptr = crtTblStmt;
 321         sprintf(ptr, "CREATE TABLE %s ( ", tableName);
 322         ptr += strlen(ptr);
 323         while (icol <= totalFields) {
 324             retValue = SQLDescribeCol(hstmt, icol, colName, colNameMax,
 325                                         &nameLength, &colType, &colLength,
 326                                         &scale, &nullable);
 327             if (retValue) {
 328                 printError(ErrSysInit, "Unable to retrieve ODBC column info\n");
 329                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 330                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 331                 SQLDisconnect (hdbc);
 332                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 333                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 334                 delete inf;
 335                 return ErrSysInit;
 336             }
 337             Util::str_tolower((char*)colName);
 338             printDebug(DM_Gateway, "Describe Column %s %d %d \n", colName, colType, colLength);
 339             icol++;
 340             if(strcmp((char*)colName,fieldName)== 0)
 341             {
 342                 isKeyFld=true;
 343                 isNullfld=true;
 344             }
 345             bool isPriFld=false;
 346             if (nullable) {
 347                 inf->list.resetIter();
 348                 while ((name=inf->list.nextFieldName())!=NULL) {
 349                     if(0==strcmp((char*)colName,name)) {
 350                         if (firstFld) {
 351                             firstFld = false;
 352                             sprintf(ptr, "%s %s", colName, AllDataType::getSQLString(AllDataType::convertFromSQLType(colType,colLength,scale,tdbName)));
 353                             ptr += strlen(ptr);
 354                             if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY)
 355                             {
 356                                 sprintf(ptr, "(%d) NOT NULL",colLength);
 357                             } else { sprintf(ptr, " NOT NULL"); }
 358                                 ptr += strlen(ptr);
 359                         } else {
 360                             sprintf(ptr, ", %s %s", colName, AllDataType::getSQLString(AllDataType::convertFromSQLType(colType,colLength,scale,tdbName)));
 361                             ptr += strlen(ptr);
 362                             if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY)
 363                             {
 364                                 sprintf(ptr, "(%d) NOT NULL",colLength);
 365                             } else { sprintf(ptr, " NOT NULL"); }
 366                             ptr += strlen(ptr);
 367                         }
 368                         tabDef.addField((char*)colName, AllDataType::convertFromSQLType(colType,colLength,scale,tdbName), colLength +1, NULL, true);
 369                         isPriFld=true;
 370                         break;
 371                     }
 372                 }
 373                 if(!isPriFld) {
 374                     if(!isNullfld) {
 375                         if (firstFld) {
 376                             firstFld = false;
 377                             sprintf(ptr, "%s %s", colName, AllDataType::getSQLString(AllDataType::convertFromSQLType(colType,colLength,scale,tdbName)));
 378                             ptr += strlen(ptr);
 379                             if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
 380                                 sprintf(ptr, "(%d)",colLength);
 381                                 ptr += strlen(ptr);
 382                             }
 383                         } else {
 384                             sprintf(ptr, ", %s %s", colName, AllDataType::getSQLString(AllDataType::convertFromSQLType(colType,colLength,scale,tdbName)));
 385                             ptr += strlen(ptr);
 386                             if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
 387                                 sprintf(ptr, "(%d)",colLength);
 388                                 ptr += strlen(ptr);
 389                             }
 390                         }
 391                         tabDef.addField((char*)colName, AllDataType::convertFromSQLType(colType,colLength,scale,tdbName), colLength+1);
 392                     } else {
 393                         if (firstFld) {
 394                             firstFld = false;
 395                             sprintf(ptr, "%s %s", colName, AllDataType::getSQLString(AllDataType::convertFromSQLType(colType,colLength,scale,tdbName)));
 396                             ptr += strlen(ptr);
 397                             if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
 398                                 sprintf(ptr, "(%d) NOT NULL",colLength);
 399                             } else { sprintf(ptr, " NOT NULL",colLength); }
 400                             ptr += strlen(ptr);
 401                         } else {
 402                             sprintf(ptr, ", %s %s", colName, AllDataType::getSQLString(AllDataType::convertFromSQLType(colType,colLength,scale,tdbName)));
 403                             ptr += strlen(ptr);
 404                             if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
 405                                 sprintf(ptr, "(%d) NOT NULL",colLength);
 406                             } else { sprintf(ptr, " NOT NULL",colLength); }
 407                             ptr += strlen(ptr);
 408                         }
 409                         tabDef.addField((char*)colName, AllDataType::convertFromSQLType(colType,colLength,scale,tdbName), colLength+1, NULL, true);
 410                         isNullfld=false;
 411                     }
 412                 }
 413             } else {
 414                 if (firstFld) {
 415                     firstFld = false;
 416                     sprintf(ptr, "%s %s", colName, AllDataType::getSQLString(AllDataType::convertFromSQLType(colType,colLength,scale,tdbName)));
 417                     ptr += strlen(ptr);
 418                     if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
 419                         sprintf(ptr, "(%d) NOT NULL",colLength);
 420                     } else { sprintf(ptr, " NOT NULL",colLength); }
 421                     ptr += strlen(ptr);
 422                 } else {
 423                     sprintf(ptr, ", %s %s", colName, AllDataType::getSQLString(AllDataType::convertFromSQLType(colType,colLength, scale, tdbName)));
 424                     ptr += strlen(ptr);
 425                     if (colType == SQL_CHAR || colType == SQL_VARCHAR || colType == SQL_BINARY) {
 426                         sprintf(ptr, "(%d) NOT NULL",colLength);
 427                     } else { sprintf(ptr, " NOT NULL",colLength); }
 428                     ptr += strlen(ptr);
 429                 }
 430                 tabDef.addField((char*)colName, AllDataType::convertFromSQLType(colType,colLength,scale, tdbName), colLength +1, NULL, true);
 431             }
 432         }
 433         sprintf(ptr, ");");
 434         ptr += strlen(ptr);
 435         //printf("table stmt '%s'\n", crtTblStmt);
 436         if(((strcmp(fieldName,"")!=0) && (strcmp(fieldName,"NULL")!=0))
 437                                                                && !isKeyFld) {
 438             printError(ErrSysInit, "Unable to cache Table for %s with key field %s\n", tableName,fieldName);
 439             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 440             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 441             SQLDisconnect (hdbc);
 442             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 443             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 444             delete inf;
 445             return ErrSysInit;
 446         }
 447         rv = stmt->prepare(crtTblStmt);
 448         if (rv != OK) {
 449             printError(ErrSysInit, "Unable to prepare create table stmt\n");
 450             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 451             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 452             SQLDisconnect (hdbc);
 453             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 454             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 455             delete inf;
 456             return ErrSysInit;
 457         }
 458         int rows = 0;
 459         rv = stmt->execute(rows);
 460         if (rv != OK) {
 461             printError(ErrSysInit, "Unable to execute create table stmt\n");
 462             SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 463             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 464             SQLDisconnect (hdbc);
 465             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 466             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 467             delete inf;
 468             return ErrSysInit;
 469         }
 470         //printf("Table created from create table stmt\n");
 471
 472         //Table is created.
 473         //Create primary key index if present
 474         if (isPriIndex && ( iskeyfieldExist ||
 475            (strcmp(fieldlistVal,"")==0  || strcmp(fieldlistVal,"NULL")== 0))) {
 476             rv = stmt->prepare(crtIdxStmt);
 477             if (rv != OK) {
 478                 printError(ErrSysInit, "Unable to prepare create table stmt\n");
 479                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 480                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 481                 SQLDisconnect (hdbc);
 482                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 483                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 484                 delete inf;
 485                 return ErrSysInit;
 486             }
 487             int rows = 0;
 488             rv = stmt->execute(rows);
 489             if (rv != OK) {
 490                 printError(ErrSysInit, "Unable to execute create table stmt\n");
 491                 SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
 492                 SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 493                 SQLDisconnect (hdbc);
 494                 SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 495                 SQLFreeHandle (SQL_HANDLE_ENV, henv);
 496                 delete inf;
 497                 return ErrSysInit;
 498             }
 499             //printf("Primary index created from create Index stmt\n");
 500         }
 501         retValue = SQLCloseCursor(hstmtmeta);
 502         rv = createIndex(hstmtmeta, tableName, inf, stmt,isPKFieldSpecified);
 503         if(rv!=OK) {
 504             dbMgr->dropTable(tableName);
 505             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 506             SQLDisconnect (hdbc);
 507             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 508             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 509             //delete inf;
 510             return rv;
 511         }
 512         //delete inf;
 513     }
 514     // Now load the table with records
 515     char insStmt[1024];
 516     char *ptr = insStmt;
 517     sprintf(ptr,"INSERT INTO %s VALUES(", tableName);
 518     ptr += strlen(ptr);
 519     bool firstFld = true;
 520     SqlStatement *sqlStmt = (SqlStatement *)stmt->getInnerStatement();
 521     sqlStmt->setConnection(con);
 522     List fNameList = sqlStmt->getFieldNameList(tableName);
 523     int noOfFields = fNameList.size();
 524
 525     while (noOfFields--) {
 526         if (firstFld) {
 527             firstFld = false;
 528             sprintf(ptr,"?", tableName);
 529             ptr += strlen(ptr);
 530         } else {
 531             sprintf(ptr, ",?");
 532             ptr += strlen(ptr);
 533         }
 534     }
 535     sprintf(ptr, ");");
 536     ptr += strlen(ptr);
 537     //printf("insert stmt: '%s'\n", insStmt);
 538
 539     rv = stmt->prepare(insStmt);
 540     if (rv != OK) {
 541         printError(ErrSysInit, "Unable to prepare create table stmt\n");
 542         SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 543         SQLDisconnect (hdbc);
 544         SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 545         SQLFreeHandle (SQL_HANDLE_ENV, henv);
 546         return ErrSysInit;
 547     }
 548     ListIterator fNameIter = fNameList.getIterator();
 549     FieldInfo *info = new FieldInfo();
 550     int fcount =1; void *valBuf;
 551     Identifier *elem = NULL;
 552     void *tembuf=NULL;//For postgre BigInt type
 553     BindBuffer *bBuf;
 554     List valBufList;
 555     SQLINTEGER len[IDENTIFIER_LENGTH];
 556     while (fNameIter.hasElement()) {
 557         elem = (Identifier*) fNameIter.nextElement();
 558         sqlStmt->getFieldInfo(tableName, (const char*)elem->name, info);
 559         if( info->type == typeString)
 560             valBuf = AllDataType::alloc(info->type, info->length+1);
 561         else
 562             valBuf = AllDataType::alloc(info->type, info->length);
 563         os::memset(valBuf,0,info->length);
 564         switch(info->type)
 565         {
 566             case typeDate:
 567                 bBuf = new BindBuffer();
 568                 bBuf->csql = valBuf;
 569                 bBuf->type = typeDate;
 570                 bBuf->length = sizeof(DATE_STRUCT);
 571                 bBuf->targetdb = malloc(bBuf->length);
 572                 memset(bBuf->targetdb, 0, bBuf->length);
 573                 valBuf = bBuf->targetdb;
 574                 valBufList.append(bBuf);
 575                 break;
 576             case typeTime:
 577                 bBuf = new BindBuffer();
 578                 bBuf->csql = valBuf;
 579                 bBuf->type = typeTime;
 580                 bBuf->length = sizeof(TIME_STRUCT);
 581                 bBuf->targetdb = malloc(bBuf->length);
 582                 memset(bBuf->targetdb, 0, bBuf->length);
 583                 valBuf = bBuf->targetdb;
 584                 valBufList.append(bBuf);
 585                 break;
 586             case typeTimeStamp:
 587                 bBuf = new BindBuffer();
 588                 bBuf->csql = valBuf;
 589                 bBuf->type = typeTimeStamp;
 590                 bBuf->length = sizeof(TIMESTAMP_STRUCT);
 591                 bBuf->targetdb = malloc(bBuf->length);
 592                 memset(bBuf->targetdb, 0, bBuf->length);
 593                 valBuf = bBuf->targetdb;
 594                 valBufList.append(bBuf);
 595                 break;
 596             case typeLongLong:
 597             {
 598                 if( tdbName == postgres )
 599                 {
 600                     bBuf = new BindBuffer();
 601                     bBuf->type = typeLongLong;
 602                     bBuf->length = 40;
 603                     bBuf->csql = valBuf;
 604                     bBuf->targetdb = AllDataType::alloc(typeString,bBuf->length);
 605                     memset(bBuf->targetdb, 0, bBuf->length);
 606                     valBuf = bBuf->targetdb;
 607                     valBufList.append(bBuf);
 608                     break;
 609                 }
 610                 else
 611                 {
 612                     bBuf = new BindBuffer();
 613                     bBuf->type = info->type;
 614                     bBuf->csql = valBuf;
 615                     valBufList.append(bBuf);
 616                     bBuf->length = info->length;
 617                     break;
 618                 }
 619             }
 620             case typeString:
 621                 if( tdbName != mysql)
 622                 {
 623                     bBuf = new BindBuffer();
 624                     bBuf->type = typeString;
 625                     bBuf->csql = valBuf;
 626                     bBuf->length = info->length+1;
 627                     valBufList.append(bBuf);
 628                     break;
 629                 }
 630             default:
 631                 bBuf = new BindBuffer();
 632                 bBuf->type = info->type;
 633                 bBuf->csql = valBuf;
 634                 valBufList.append(bBuf);
 635                 bBuf->length = info->length;
 636                 break;
 637         }
 638         //os::memset(valBuf,0,bBuf->length);
 639         retValue = SQLBindCol (hstmt, fcount, AllDataType::convertToSQL_C_Type(info->type,tdbName), valBuf, bBuf->length, &len[fcount]);
 640         fcount++;
 641         if (retValue) {
 642             printError(ErrSysInit, "Unable to bind columns in ODBC\n");
 643             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 644             SQLDisconnect (hdbc);
 645             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 646             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 647             return ErrSysInit;
 648         }
 649     }
 650     delete info;
 651     fNameIter.reset();
 652     while (fNameIter.hasElement())
 653         delete ((FieldName *) fNameIter.nextElement());
 654     fNameList.reset();
 655
 656     retValue = SQLExecute (hstmt);
 657     if (retValue) {
 658         printError(ErrSysInit, "Unable to execute ODBC statement\n");
 659         SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 660         SQLDisconnect (hdbc);
 661         SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 662         SQLFreeHandle (SQL_HANDLE_ENV, henv);
 663         return ErrSysInit;
 664     }
 665     int fldpos=0;
 666     int countForCommit = 0;
 667     while(true) {
 668         //TODO: if SQLFetch return other than record not found error
 669         //it should drop the table
 670         retValue = SQLFetch (hstmt);
 671         if (retValue) break;
 672         ListIterator bindIter = valBufList.getIterator();
 673         fldpos = 0;
 674         while (bindIter.hasElement()) {
 675             bBuf = (BindBuffer*) bindIter.nextElement();
 676             switch (bBuf->type) {
 677                 case typeString:
 678                 {
 679                     if( tdbName != mysql)
 680                     {
 681                         Util::trimRight((char*)bBuf->csql);
 682                     }
 683                     break;
 684                 }
 685                 case typeDate:
 686                 {
 687                     Date *dtCSQL = (Date*) bBuf->csql;
 688                     DATE_STRUCT *dtTarget = (DATE_STRUCT*) bBuf->targetdb;
 689                     dtCSQL->set(dtTarget->year,dtTarget->month,dtTarget->day);
 690                     break;
 691                 }
 692                 case typeTime:
 693                 {
 694                     Time *dtCSQL = (Time*) bBuf->csql;
 695                     TIME_STRUCT *dtTarget = (TIME_STRUCT*) bBuf->targetdb;
 696                     dtCSQL->set(dtTarget->hour,dtTarget->minute,dtTarget->second);
 697                     break;
 698                 }
 699                 case typeTimeStamp:
 700                 {
 701                     TimeStamp *dtCSQL = (TimeStamp*) bBuf->csql;
 702                     TIMESTAMP_STRUCT *dtTarget = (TIMESTAMP_STRUCT*) bBuf->targetdb;
 703                     dtCSQL->setDate(dtTarget->year,dtTarget->month,dtTarget->day);
 704                     dtCSQL->setTime(dtTarget->hour,dtTarget->minute,dtTarget->second, dtTarget->fraction);
 705                     break;
 706                 }
 707                 case typeLongLong:
 708                 {
 709                     if ( tdbName == postgres) {
 710                         sscanf((const char*)bBuf->targetdb,"%lld",(long long*) bBuf->csql);
 711                     }
 712                     break;
 713                 }
 714             }
 715             setParamValues(stmt, ++fldpos, bBuf->type, bBuf->length, (char *) bBuf->csql);
 716         }
 717         fldpos=0;
 718         //table->resetNullinfo();
 719         while(fldpos < fcount-1) {
 720             if(len[++fldpos] == SQL_NULL_DATA) {
 721                 stmt->setNull(fldpos);
 722             }
 723         }
 724         int rows = 0;
 725         rv = stmt->execute(rows);
 726         if (rv != OK) {
 727             printError(ErrSysInit, "Unable to cache record in CSQL.\n");
 728             SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 729             SQLDisconnect (hdbc);
 730             SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 731             SQLFreeHandle (SQL_HANDLE_ENV, henv);
 732             return ErrSysInit;
 733         }
 734         countForCommit++;
 735         if (countForCommit == 1000) {
 736             countForCommit = 0;
 737             conn->commit();
 738             conn->beginTrans();
 739         }
 740     }
 741     //TODO::leak:: valBufList and its targetdb buffer
 742     ListIterator it = valBufList.getIterator();
 743     while(it.hasElement()) {
 744         BindBuffer *bb = (BindBuffer *) it.nextElement();
 745         if (bb->csql) { free(bb->csql); bb->csql = NULL; }
 746         if (bb->targetdb) { free(bb->targetdb); bb->targetdb = NULL; }
 747         delete bb; bb = NULL;
 748     }
 749     valBufList.reset();
 750     SQLCloseCursor (hstmt);
 751     SQLFreeHandle (SQL_HANDLE_STMT, hstmt);
 752     SQLDisconnect (hdbc);
 753     SQLFreeHandle (SQL_HANDLE_DBC, hdbc);
 754     SQLFreeHandle (SQL_HANDLE_ENV, henv);
 755     return OK;
 756 }
 757
 758 DbRetVal CacheTableLoader::reload()
 759 {
 760     FILE *fp=NULL;
 761     DbRetVal rv = unload(false);
 762     if (rv != OK) return rv;
 763     //get table cache senarios
 764     fp = fopen(Conf::config.getTableConfigFile(),"r");
 765     if( fp == NULL ) {
 766         printError(ErrSysInit, "cachetable.conf file does not exist");
 767         return OK;
 768     }
 769     int mode;
 770     rv = OK;
 771     char tablename[IDENTIFIER_LENGTH];
 772     char fieldname[IDENTIFIER_LENGTH];
 773     char field[IDENTIFIER_LENGTH];
 774     char condition[IDENTIFIER_LENGTH];
 775     char dsnname[IDENTIFIER_LENGTH];
 776     while(!feof(fp))
 777     {
 778         fscanf(fp, "%d %s %s %s %s %s\n", &mode, tablename,fieldname,condition,field,dsnname);
 779         if(strcmp(tablename,tableName)==0) break;
 780     }
 781     fclose(fp);
 782     setCondition(TableConf::config.getRealConditionFromFile(condition));
 783     setFieldName(fieldname);
 784     setFieldListVal(field);
 785     setDsnName(dsnname);
 786     rv = load(false);
 787     return rv;
 788 }
 789
 790 DbRetVal CacheTableLoader::unload(bool tabDefinition)
 791 {
 792     AbsSqlConnection *conn = SqlFactory::createConnection(CSqlLog);
 793     DbRetVal rv = conn->connect(userName, password);
 794     if (rv != OK) return ErrSysInit;
 795     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlLog);
 796     stmt->setConnection(conn);
 797     SqlLogConnection *logConn = (SqlLogConnection *) conn;
 798     logConn->setNoMsgLog(true);
 799     char statement[1024];
 800     if (TableConf::config.isTableCached(tableName) != OK) {
 801         printError(ErrNotCached, "The table \'%s\' is not cached", tableName);
 802         conn->disconnect();
 803         delete stmt;
 804         delete conn;
 805         return ErrNotCached;
 806     }
 807     SqlConnection *con = (SqlConnection *) conn->getInnerConnection();
 808     DatabaseManager *dbMgr = (DatabaseManager*) con->getConnObject().getDatabaseManager();
 809     if (dbMgr == NULL) {
 810         conn->disconnect();
 811         delete stmt; delete conn;
 812         printError(ErrSysInit, "Auth failed\n");
 813         return ErrSysInit;
 814     }
 815     if (!tabDefinition)
 816     {
 817         sprintf(statement, "DELETE FROM %s;", tableName);
 818         SqlStatement *sqlStmt = (SqlStatement*)stmt;
 819         sqlStmt->setLoading(true);
 820         rv = stmt->prepare(statement);
 821         if (rv != OK) {
 822             conn->disconnect();
 823             delete stmt; delete conn;
 824             return ErrBadCall;
 825         }
 826         conn->beginTrans();
 827         int rows = 0;
 828         rv = stmt->execute(rows);
 829         if (rv != OK) {
 830             conn->disconnect();
 831             delete stmt; delete conn;
 832             return ErrBadCall;
 833         }
 834         conn->commit();
 835     }
 836     else
 837     {
 838         rv = TableConf::config.removeFromCacheTableFile();
 839         if (rv != OK) {
 840             conn->disconnect(); delete stmt; delete conn;
 841             return ErrBadCall;
 842         }
 843         sprintf(statement, "DROP TABLE %s;", tableName);
 844         SqlStatement *sqlStmt = (SqlStatement*)stmt;
 845         sqlStmt->setLoading(true);
 846         rv = stmt->prepare(statement);
 847         if (rv != OK) {
 848             //TableConf::config.addToCacheTableFile(false);
 849             conn->disconnect();
 850             delete stmt; delete conn;
 851             return ErrBadCall;
 852         }
 853         int rows = 0;
 854         rv = stmt->execute(rows);
 855         if (rv != OK) {
 856             //TableConf::config.addToCacheTableFile(false);
 857             conn->disconnect(); delete stmt; delete conn;
 858             return ErrBadCall;
 859         }
 860     }
 861     conn->disconnect();
 862     delete stmt; delete conn;
 863     return rv;
 864 }
 865
 866 DbRetVal CacheTableLoader::refresh()
 867 {
 868     return OK;
 869 }
 870
 871 DbRetVal CacheTableLoader::recoverAllCachedTables()
 872 {
 873     FILE *fp;
 874     Connection conn;
 875     DbRetVal rv = conn.open(userName, password);
 876     if(rv !=OK) return ErrSysInit;
 877
 878     //Note: if connection is not open, configuration veriables may be incorrect
 879
 880     fp = fopen(Conf::config.getTableConfigFile(),"r");
 881     if( fp == NULL ) {
 882         printError(ErrSysInit, "cachetable.conf file does not exist");
 883         conn.close();
 884         return OK;
 885     }
 886     conn.close();
 887     //TODO::take exclusive lock on database
 888     char tablename[IDENTIFIER_LENGTH];
 889     char fieldname[IDENTIFIER_LENGTH];
 890     char condition[IDENTIFIER_LENGTH];
 891     char field[IDENTIFIER_LENGTH];
 892     char dsnname[IDENTIFIER_LENGTH];
 893
 894     int mode;
 895     int scanItems=0;
 896     rv = OK;
 897     while(!feof(fp))
 898     {
 899         scanItems = fscanf(fp, "%d %s %s %s %s %s\n", &mode, tablename,fieldname,condition,field,dsnname);
 900         if (scanItems != 6) {
 901             tablename[0]='\0';
 902             printf("There is no table to be cached.\n");
 903             return OK;
 904         }
 905         //if (mode ==2 )  //just replicated table and not cached
 906         //continue;
 907         printDebug(DM_Gateway, "Recovering Table from target db: %s\n", tablename);
 908         setCondition(TableConf::config.getRealConditionFromFile(condition));
 909          if( (strcmp(Conf::config.getDSN(),dsnname)!=0) ){
 910                      setDsnName(dsnname);
 911              setTable(tablename);
 912              setFieldName(fieldname);
 913              setFieldListVal(field);
 914              printf("Recovering table %s %s %s\n", tablename,condition,field);
 915              rv = load();
 916              if (rv != OK) { fclose(fp); return rv; }
 917          } else {
 918              setDsnName(Conf::config.getDSN());
 919              setTable(tablename);
 920              setFieldName(fieldname);
 921              setFieldListVal(field);
 922              printf("Recovering table %s %s %s\n", tablename,condition,field);
 923              rv = load();
 924              if (rv != OK) { fclose(fp); return rv; }
 925          }
 926     }
 927     fclose(fp);
 928     return OK;
 929 }
 930
 931 void CacheTableLoader::setParamValues(AbsSqlStatement *stmt, int parampos, DataType type, int length, char *value)
 932 {
 933     switch(type)
 934     {
 935         case typeInt:
 936             stmt->setIntParam(parampos, *(int*)value);
 937             break;
 938         case typeLong:
 939             stmt->setLongParam(parampos, *(long*)value);
 940             break;
 941         case typeLongLong:
 942             stmt->setLongLongParam(parampos, *(long long*)value);
 943             break;
 944         case typeShort:
 945             stmt->setShortParam(parampos, *(short*)value);
 946             break;
 947         case typeByteInt:
 948             stmt->setByteIntParam(parampos, *(char*)value);
 949             break;
 950         case typeDouble:
 951             stmt->setDoubleParam(parampos, *(double*)value);
 952             break;
 953         case typeFloat:
 954             stmt->setFloatParam(parampos, *(float*)value);
 955             break;
 956         case typeDate:
 957             stmt->setDateParam(parampos, *(Date*)value);
 958             break;
 959         case typeTime:
 960             stmt->setTimeParam(parampos, *(Time*)value);
 961             break;
 962         case typeTimeStamp:
 963             stmt->setTimeStampParam(parampos, *(TimeStamp*)value);
 964             break;
 965         case typeString:
 966             {
 967                 char *d =(char*)value;
 968                 d[length-1] = '\0';
 969                 stmt->setStringParam(parampos, (char*)value);
 970                 break;
 971             }
 972         case typeBinary:
 973             stmt->setBinaryParam(parampos, (char *) value, length);
 974             break;
 975     }
 976     return;
 977 }
 978
 979 DbRetVal CacheTableLoader::createIndex(SQLHSTMT hstmtmeta, char *tableName, HashIndexInitInfo *inf,AbsSqlStatement *stmt,bool isPKFieldSpecified)
 980 {
 981         bool isKeyFld= false;
 982         int retValue = 0;
 983         char columnname[IDENTIFIER_LENGTH];
 984         char indexname[IDENTIFIER_LENGTH];
 985         short type;
 986         short unique;
 987         char *name = NULL;
 988         DbRetVal rv = OK;
 989         retValue = SQLStatistics(hstmtmeta, NULL, 0, NULL, SQL_NTS,
 990                     (SQLCHAR*) tableName, SQL_NTS, SQL_INDEX_ALL, SQL_QUICK);
 991         retValue = SQLBindCol(hstmtmeta, 4, SQL_C_SHORT,
 992                           &unique, 2, NULL);
 993         retValue = SQLBindCol(hstmtmeta, 6, SQL_C_CHAR,
 994                           indexname, 129, NULL);
 995         retValue = SQLBindCol(hstmtmeta, 7, SQL_C_SHORT,
 996                           &type, 2, NULL);
 997         retValue = SQLBindCol(hstmtmeta, 9, SQL_C_CHAR,
 998                           columnname, 129,NULL);
 999         List indexList;
1000         bool isSecondTime = false;
1001         CacheIndexInfo *info=NULL;
1002         while ((retValue = SQLFetch(hstmtmeta)) == SQL_SUCCESS) {
1003         //if (type != SQL_TABLE_STAT)
1004             {
1005                 printDebug(DM_Gateway, "Column: %-18s Index Name: %-18s unique:%hd type:%hd\n", columnname, indexname, unique, type);
1006             }
1007
1008             if (type == 3)
1009             {
1010
1011                 bool isFldAdd = false;
1012                 ListIterator iter = indexList.getIterator();
1013                 iter.reset();
1014                 while (iter.hasElement())
1015                 {
1016                     CacheIndexInfo *indInfo = (CacheIndexInfo *)iter.nextElement();
1017                     if(0 == strcmp( indInfo->indexName, indexname))
1018                     {
1019                          indInfo->fieldList.append(columnname);
1020                          isFldAdd = true;
1021                     }
1022                 }
1023                 if(!isFldAdd){
1024                     info = new CacheIndexInfo();
1025                     info->fieldList.append(columnname);
1026                     strcpy(info->indexName, indexname);
1027                     indexList.append(info);
1028                     isSecondTime = true;
1029                 }
1030
1031             }
1032         }
1033         ListIterator iter = indexList.getIterator();
1034         iter.reset();
1035         int noOfPkfield = inf->list.size();
1036         char *fName=NULL;
1037         char *cptr = NULL;
1038         while (iter.hasElement())
1039         {
1040             cptr = columnname;
1041             bool isFieldExistInCondition = false;
1042             bool isPrimary=false;
1043             CacheIndexInfo *indInfo = (CacheIndexInfo *)iter.nextElement();
1044             int noOfFld= indInfo->fieldList.size();
1045             indInfo->fieldList.resetIter();
1046             while ((fName = indInfo->fieldList.nextFieldName())!=NULL)
1047             {
1048                 if(( 1 == noOfFld) && (0 == strcmp(fName,fieldName))) { isKeyFld=true; }
1049                 inf->list.resetIter();
1050                 while ((name=inf->list.nextFieldName())!=NULL)
1051                 {
1052                     if(0==strcmp(fName,name)) {  isPrimary = true;  break;  }
1053                     isPrimary = false;
1054                 }
1055                 if (!TableConf::config.isFieldExist(fName) && ( (strcmp(fieldlistVal,"")!=0) && (strcmp(fieldlistVal,"NULL")!=0) ))
1056                 {
1057                      isFieldExistInCondition =true;
1058                      continue;
1059                 }
1060                 sprintf(cptr, "%s ,",fName);
1061                 cptr += strlen(cptr);
1062
1063             }
1064             if(isFieldExistInCondition) continue;
1065             cptr -=1;
1066             *cptr = '\0';
1067
1068             if (isPrimary) { continue; }
1069             char crtIdxStmt[1024];
1070             char indname[128];
1071             sprintf(indname, "%s_%s", tableName, indInfo->indexName);
1072             sprintf(crtIdxStmt, "CREATE INDEX %s on %s(%s) HASH SIZE 10007;", indname, tableName, columnname);
1073                 //printf("create index stmt \n'%s'\n", crtIdxStmt);
1074             rv = stmt->prepare(crtIdxStmt);
1075             if (rv != OK) {
1076                 printError(ErrSysInit, "Unable to prepare create table stmt\n");
1077                 return ErrSysInit;
1078             }
1079             int rows = 0;
1080             rv = stmt->execute(rows);
1081             if (rv != OK) {
1082                 printError(ErrSysInit, "Unable to execute create table stmt\n");
1083                 return ErrSysInit;
1084             }
1085             delete indInfo;
1086         }// while meta data fetch for index creation
1087         delete inf;
1088         SQLCloseCursor (hstmtmeta);
1089         SQLFreeHandle (SQL_HANDLE_STMT, hstmtmeta);
1090         if( !isKeyFld && isPKFieldSpecified) {
1091             if(shouldForce) {
1092                 char frcIndStmt[1024];
1093                 char indname[128];
1094                 sprintf(indname, "%s_%s", tableName, "keyInd");
1095                 sprintf(frcIndStmt, "CREATE INDEX %s on %s(%s) HASH;", indname, tableName, fieldName);
1096                 rv = stmt->prepare(frcIndStmt);
1097                 if (rv != OK) {
1098                     printError(ErrSysInit, "Unable to prepare create table stmt\n");
1099                     return ErrSysInit;
1100                 }
1101                 int rows = 0;
1102                 rv = stmt->execute(rows);
1103                 if (rv != OK) {
1104                     printError(ErrSysInit, "Unable to execute create table stmt\n");
1105                     return ErrSysInit;
1106                 }
1107             } else {
1108                 printError(ErrSysInit, "Unable to cache Table for %s with key field %s\n", tableName,fieldName);
1109                 return ErrSysInit;
1110             }
1111         }
1112      return OK;
1113 }
1114
1115