src/tools/redo.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 <Statement.h>
  18 #include <SqlFactory.h>
  19 #include <SqlStatement.h>
  20 #define SQL_STMT_LEN 1024
  21 #define DEBUG 1
  22
  23 AbsSqlConnection *conn;
  24 void *stmtBuckets;
  25 void addToHashTable(int stmtID, AbsSqlStatement* sHdl)
  26 {
  27     int bucketNo = stmtID % STMT_BUCKET_SIZE;
  28     StmtBucket *buck = (StmtBucket *) stmtBuckets;
  29     StmtBucket *stmtBucket = &buck[bucketNo];
  30     StmtNode *node = new StmtNode();
  31     node->stmtId = stmtID;
  32     node->stmt = sHdl;
  33     stmtBucket->bucketList.append(node);
  34     return;
  35 }
  36 void removeFromHashTable(int stmtID)
  37 {
  38     int bucketNo = stmtID % STMT_BUCKET_SIZE;
  39     StmtBucket *buck = (StmtBucket *) stmtBuckets;
  40     StmtBucket *stmtBucket = &buck[bucketNo];
  41     StmtNode *node = NULL;
  42     ListIterator it = stmtBucket->bucketList.getIterator();
  43     while(it.hasElement()) {
  44         node = (StmtNode *) it.nextElement();
  45         if(stmtID == node->stmtId) break;
  46     }
  47     it.reset();
  48     stmtBucket->bucketList.remove(node);
  49     return;
  50 }
  51 AbsSqlStatement *getStmtFromHashTable(int stmtId)
  52 {
  53     int bucketNo = stmtId % STMT_BUCKET_SIZE;
  54     StmtBucket *buck = (StmtBucket *) stmtBuckets;
  55     StmtBucket *stmtBucket = &buck[bucketNo];
  56     StmtNode *node = NULL;
  57     ListIterator it = stmtBucket->bucketList.getIterator();
  58     while(it.hasElement()) {
  59         node = (StmtNode *) it.nextElement();
  60         if(stmtId == node->stmtId) {
  61             return node->stmt;
  62         }
  63     }
  64     return NULL;
  65 }
  66 void freeAllStmtHandles()
  67 {
  68     //TODO
  69 }
  70 void setParam(AbsSqlStatement *stmt, int pos, DataType type , int length, void *value)
  71 {
  72     switch(type)
  73     {
  74         case typeInt:
  75             stmt->setIntParam(pos, *(int*)value);
  76             break;
  77         case typeLong:
  78             stmt->setLongParam(pos, *(long*) value);
  79             break;
  80         case typeLongLong:
  81             stmt->setLongLongParam(pos, *(long long*)value);
  82             break;
  83         case typeShort:
  84             stmt->setShortParam(pos, *(short*)value);
  85             break;
  86         case typeByteInt:
  87             stmt->setByteIntParam(pos, *(ByteInt*)value);
  88             break;
  89         case typeDouble:
  90             stmt->setDoubleParam(pos, *(double*)value);
  91             break;
  92         case typeFloat:
  93             stmt->setFloatParam(pos, *(float*)value);
  94             break;
  95         case typeDate:
  96             stmt->setDateParam(pos, *(Date*)value);
  97             break;
  98         case typeTime:
  99             stmt->setTimeParam(pos, *(Time*)value);
 100             break;
 101         case typeTimeStamp:
 102             stmt->setTimeStampParam(pos, *(TimeStamp*)value);
 103             break;
 104         case typeString:
 105             stmt->setStringParam(pos, (char*)value);
 106             break;
 107         case typeBinary:
 108             stmt->setBinaryParam(pos, value, length);
 109             break;
 110         default:
 111             printf("unknown type\n");
 112             break;
 113     }
 114     return;
 115 }
 116
 117 int main(int argc, char **argv)
 118 {
 119     struct stat st;
 120     char fileName[1024];
 121     int c = 0, opt=0;
 122     bool interactive=0;
 123     while ((c = getopt(argc, argv, "ai?")) != EOF) {
 124         switch (c) {
 125             case '?' : { opt = 1; break; } //print help
 126             case 'a' : { opt = 2; break; }
 127             case 'i' : { interactive = 1; break; }
 128             default: printf("Wrong args\n"); exit(1);
 129
 130         }
 131     }//while options
 132     if (2 !=opt) {
 133        printf("This is an internal csql command.");
 134        exit(1);
 135     }
 136     Conf::config.readAllValues(os::getenv("CSQL_CONFIG_FILE"));
 137     sprintf(fileName, "%s/csql.db.cur", Conf::config.getDbFile());
 138     printf("Redo log filename is :%s\n", fileName);
 139     int fd = open(fileName, O_RDONLY);
 140     if (-1 == fd) { return OK; }
 141     if (fstat(fd, &st) == -1) {
 142         printf("Unable to retrieve undo log file size\n");
 143         close(fd);
 144         return ErrUnknown;
 145     }
 146     if (st.st_size ==0) {
 147         printf("Nothing in redo log file\n");
 148         close(fd);
 149         return 0;
 150     }
 151     conn = SqlFactory::createConnection(CSqlDirect);
 152     DbRetVal rv = conn->connect(I_USER, I_PASS);
 153     SqlConnection *sCon = (SqlConnection*) conn;
 154     rv = sCon->getExclusiveLock();
 155     if (rv != OK) {
 156         close(fd);
 157         conn->disconnect();
 158         delete conn;
 159         return 1;
 160     }
 161     void *startAddr = mmap(NULL, st.st_size, PROT_READ, MAP_PRIVATE, fd, 0);
 162     if (MAP_FAILED == startAddr) {
 163         printf("Unable to read undo log file:mmap failed.\n");
 164         conn->disconnect();
 165         delete conn;
 166         return 2;
 167     }
 168     stmtBuckets = malloc (STMT_BUCKET_SIZE * sizeof(StmtBucket));
 169     memset(stmtBuckets, 0, STMT_BUCKET_SIZE * sizeof(StmtBucket));
 170
 171     char *iter = (char*)startAddr;
 172     void *value;
 173     int logType, eType;
 174     int stmtID;
 175     int txnID;
 176     int len, ret;
 177     int loglen;
 178     char stmtString[1024];
 179     //printf("size of file %d\n", st.st_size);
 180     while(true) {
 181         //printf("OFFSET HERE %d\n", iter - (char*)startAddr);
 182         if (iter - (char*)startAddr >= st.st_size) break;
 183         logType = *(int*)iter;
 184         if (logType == -1) { //prepare
 185             iter = iter + sizeof(int);
 186             txnID = *(int*) iter; iter += sizeof(int);
 187             loglen = *(int*) iter; iter += sizeof(int);
 188             stmtID = *(int*)iter;
 189             iter = iter + sizeof(int);
 190             len = *(int*)iter;
 191             iter = iter + sizeof(int);
 192             strncpy(stmtString, iter, len);
 193             iter = iter + len;
 194             //printf("PREPARE:%d %d %s\n", stmtID, len, stmtString);
 195             AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlDirect);
 196             stmt->setConnection(conn);
 197             if (interactive) printf("PREPARE %d : %s\n", stmtID, stmtString);
 198             rv = stmt->prepare(stmtString);
 199             if (rv != OK) {
 200                 printf("unable to prepare\n");
 201                 conn->disconnect();
 202                 return ErrSysFatal;
 203             }
 204             stmt->prepare(stmtString);
 205             SqlStatement *sqlStmt = (SqlStatement*)stmt;
 206             sqlStmt->setLoading(true);
 207             addToHashTable(stmtID, stmt);
 208         }
 209         else if(logType == -2) { //commit
 210             conn->beginTrans();
 211             iter = iter + sizeof(int);
 212             txnID = *(int*) iter; iter += sizeof(int);
 213             loglen = *(int*) iter; iter += sizeof(int);
 214             char *curPtr = iter;
 215             while(true) {
 216                 //printf("Iter length %d\n", iter - curPtr);
 217                 if (iter - (char*)startAddr >= st.st_size) {
 218                     //file end reached
 219                     //printf("Redo log file end\n");
 220                     conn->commit();
 221                     freeAllStmtHandles();
 222                     conn->disconnect();
 223                     munmap((char*)startAddr, st.st_size);
 224                     close(fd);
 225                     delete conn;
 226                     return 0;
 227                 }
 228                 stmtID = *(int*)iter;
 229                 //printf("stmtid %d\n", stmtID);
 230                 if (interactive) printf("EXEC %d :\n", stmtID);
 231                 iter = iter + sizeof(int);
 232                 eType = *(int*)iter;
 233                 //printf("eType is %d\n", eType);
 234                 AbsSqlStatement *stmt = getStmtFromHashTable(stmtID);
 235                 if (0 == eType) { //execute type
 236                     iter = iter + sizeof(int);
 237                     //printf("EXEC: %d\n", stmtID);
 238                     if (stmt) {
 239                         rv = stmt->execute(ret);
 240                         if (rv != OK) {
 241                             //printf("execute failed\n");
 242                             conn->disconnect();
 243                             return ErrSysFatal;
 244                         }
 245                     } else {
 246                         printf("statement not found for %d\n",stmtID);
 247                     }
 248                     if (*(int*)iter <0) break;
 249                 } else if ( 1 == eType) { //set type
 250                     iter=iter+sizeof(int);
 251                     int pos = *(int*) iter;
 252                     iter=iter+sizeof(int);
 253                     DataType type = (DataType)(*(int*)iter);
 254                     iter=iter+sizeof(int);
 255                     int len = *(int*) iter;
 256                     iter=iter+sizeof(int);
 257                     value = iter;
 258                     //AllDataType::printVal(value, type, len);
 259                     iter=iter+len;
 260                     //printf("SET: %d %d %d %d\n", stmtID, pos, type, len);
 261                     setParam(stmt, pos, type, len, value);
 262                     if (*(int*)iter <0) break;
 263                 }
 264             }
 265             conn->commit();
 266         }
 267         else if(logType == -3) { //free
 268             iter = iter + sizeof(int);
 269             txnID = *(int*) iter; iter += sizeof(int);
 270             loglen = *(int*) iter; iter += sizeof(int);
 271             stmtID = *(int*)iter;
 272             iter = iter + sizeof(int);
 273             if (interactive) printf("FREE %d:\n", stmtID);
 274             AbsSqlStatement *stmt = getStmtFromHashTable(stmtID);
 275             if (stmt) {
 276                 stmt->free();
 277                 removeFromHashTable(stmtID);
 278             } else { printf("statement not found for %d\n",stmtID);}
 279         }
 280         else if(logType == -4) { //prepare and execute
 281             iter = iter + sizeof(int);
 282             txnID = *(int*) iter; iter += sizeof(int);
 283             loglen = *(int*) iter; iter += sizeof(int);
 284             stmtID = *(int*)iter;
 285             iter = iter + sizeof(int);
 286             len = *(int*)iter;
 287             iter = iter + sizeof(int);
 288             strncpy(stmtString, iter, len);
 289             stmtString[len+1] ='\0';
 290             iter = iter + len;
 291             //printf("CREATE:%d %d %s\n", stmtID, len, stmtString);
 292             AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlDirect);
 293             if ( NULL == stmt) {
 294                 printf("unable to prepare\n");
 295                 conn->disconnect();
 296                 return ErrSysFatal;
 297             }
 298             stmt->setConnection(conn);
 299             if (interactive) printf("EXECDIRECT %d : %s\n", stmtID, stmtString);
 300             rv = stmt->prepare(stmtString);
 301             if (rv != OK) {
 302                 printf("unable to prepare\n");
 303                 conn->disconnect();
 304                 return ErrSysFatal;
 305             }
 306             rv = stmt->execute(ret);
 307             if (rv != OK) {
 308                 if (strlen(stmtString) > 6 &&
 309                     ( (strncasecmp(stmtString,"CREATE", 6) == 0) ||
 310                       (strncasecmp(stmtString,"DROP", 4) == 0)) ) {
 311             //        conn->disconnect();
 312               //      return OK;
 313                     continue;
 314                 }
 315                 printf("unable to execute\n");
 316                 conn->disconnect();
 317                 return ErrSysFatal;
 318             }
 319             stmt->free();
 320         }
 321     }
 322     munmap((char*)startAddr, st.st_size);
 323     close(fd);
 324     freeAllStmtHandles();
 325     conn->disconnect();
 326     delete conn;
 327     return 0;
 328 }