///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoRecoveryManager.cc
// ----------------------
// Cego recovery manager class implementation
//                                                         
// Design and Implementation by Bjoern Lemke               
//
// (C)opyright 2000-2012 Bjoern Lemke
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2, or (at your option)
// any later version.
// 
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; see the file COPYING.  If not, write to
// the Free Software Foundation, 59 Temple Place - Suite 330,
// Boston, MA 02111-1307, USA.
//
// IMPLEMENTATION MODULE
//
// Class: CegoRecoveryManager
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

// base includes
#include <lfc/SetT.h>
#include <lfc/Datetime.h>
#include <lfc/Sleeper.h>
#include <lfc/Host.h>
#include <lfc/CommandExecuter.h>

// cego includes
#include "CegoXMLdef.h"
#include "CegoRecoveryManager.h"

#include <string.h>
#include <stdlib.h>

CegoRecoveryManager::CegoRecoveryManager(CegoDistManager *pGTM, CegoRecoveryManager::RecoveryMode mode)
{
    _pGTM = pGTM;
    _pLogger = _pGTM->getDBMng();
    _pDBMng = _pGTM->getDBMng();
    _recoveryMode = mode;
    _modId = _pDBMng->getModId("CegoRecoveryManager"); 	    
}

CegoRecoveryManager::~CegoRecoveryManager()
{
}

long CegoRecoveryManager::recoverTableSet(const Chain& tableSet, int pit)
{   

    _pLogger->log(_modId, Logger::NOTICE, Chain("Recovering tableset ") + tableSet + Chain(" ..."));

    char *pShell = getenv(CGEXESHELLVARNAME);
    if ( pShell == NULL )
    {
	_shellCmd = Chain(CGSTDEXESHELL);
    }
    else
    {
	_shellCmd = Chain(pShell);
    }
    
    int tabSetId = _pDBMng->getTabSetId(tableSet);

    _pDBMng->setTableSetRunState(tableSet, XML_RECOVERY_VALUE);

    _pDBMng->setRecoveryMode(tabSetId, CegoDatabaseManager::ON ); 

    try
    {
	
	while ( _pDBMng->getTableSetSyncState(tableSet) != Chain(XML_SYNCHED_VALUE) &&  _pDBMng->getRecoveryMode(tabSetId) == CegoDatabaseManager::ON)
	{
#ifdef CGDEBUG	
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Waiting for tableset sync ... "));
#endif
	    
	    Sleeper s;
	    s.secSleep(LOGMNG_RECOVERY_DELAY);
	}
	
	if ( _pDBMng->getTableSetSyncState(tableSet) != Chain(XML_SYNCHED_VALUE) )
	{
	    Chain msg = Chain("No sync on tableset ") + tableSet + Chain(", recovery failed");
	    _pDBMng->setRecoveryMode(tabSetId, CegoDatabaseManager::OFF );
	    _pDBMng->log(_modId, Logger::LOGERR, msg);
	    throw Exception(EXLOC, msg); 
	}
	
	_pGTM->initLock(tabSetId);
	_pGTM->regDataFiles(tableSet);
	
	// release logfiles to force archiving of current online log
	if ( _recoveryMode == LOCAL )
	    _pDBMng->releaseLogFiles(tableSet, true);


	// during recovery, we do not write to online log files
	_pDBMng->stopLog(tabSetId);	
	
	
	long cplsn;
	
	Chain tsTicketName = _pDBMng->getTSTicket(tableSet);
	
	File tsTicket(tsTicketName);
	if ( tsTicket.exists() ) 
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Backup tableset ticket detected, datafile file recovery required ..."));
	    tsTicket.open(File::READ);
	    
	    XMLSuite xml;
	    
	    Document *pDoc = new Document;
	    pDoc->setAttribute(XML_VERSION_ATTR, XML_VERSION_VALUE);
	    
	    xml.setDocument(pDoc);    
	    xml.setFile(&tsTicket);
	    
	    xml.parse();
	    
	    Element *pRoot = pDoc->getRootElement();
	    pDoc->setRootElement(0);
	    delete pDoc;
	    
	    _pDBMng->setTableSetInfo(tableSet, pRoot);
	    
	    cplsn = _pDBMng->getCommittedLSN(tableSet);
	    
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Starting datafile recovery for tableset ") + tableSet + Chain(" ..."));
	    long lsn = dataFileRecovery(tableSet, tabSetId, cplsn);
	    

	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Datafile recovery for tableset ") + tableSet + Chain(" finished with lsn ") + Chain(lsn));
	    
	    tsTicket.close();
	    
#ifdef CGDEBUG	
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Removing backup ticket ") + tsTicket.getFileName() + Chain(" ..."));
#endif
	    
	    tsTicket.remove();
	    
	}
	else
	{	
	    cplsn = _pDBMng->getCommittedLSN(tableSet);
	}
	
	if ( pit == 0 )
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Starting up-to-crash transaction recovery for tableset ") + tableSet + Chain(" ..."));
	else
	{
	    Datetime d(pit);
	    _pDBMng->log(_modId, Logger::NOTICE, 
			 Chain("Starting point-in-time transaction recovery to ") + d.asChain() + Chain(" for tableset ") + tableSet + Chain(" ..."));
	    
	}
	
	_pDBMng->log(_modId, Logger::NOTICE, Chain("Commited lsn =  ") + Chain(cplsn));
	
	long lsn = transactionRecovery(tableSet, tabSetId, cplsn, pit);
	
#ifdef CGDEBUG	
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Transaction recovery finished with lsn = ") + Chain(lsn));
#endif
	
	_pDBMng->initLogFiles(tableSet, true);
	
	_pDBMng->setActiveLogFile(tableSet);    
	_pDBMng->setLSN(tabSetId, lsn + 1);
	_pDBMng->startLog(tabSetId);
        
	CegoLogRecord logRec;    
	CegoLogRecord lr;
	lr.setAction(CegoLogRecord::LOGREC_SYNC);    
	if ( _pDBMng->logAction(tabSetId, lr) == CegoLogManager::LOG_ERROR )
	{
	    throw Exception(EXLOC, "Cannot write sync point after tableset recovery");
	}
	
	_pDBMng->setTableSetRunState(tableSet, XML_ONLINE_VALUE);    
	
	_pDBMng->setRecoveryMode(tabSetId, CegoDatabaseManager::OFF );
	
	long newcplsn = _pDBMng->getLSN(tabSetId) - 1;
	_pDBMng->setCommittedLSN(tabSetId, newcplsn);
	
#ifdef CGDEBUG	
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Writing sync point ..."));
#endif
	
	_pGTM->registerObjects(tableSet);
	
	_pGTM->writeCheckPoint(tableSet, true, false);

	_pDBMng->log(_modId, Logger::NOTICE, Chain("Recovery for tableset ") + tableSet + Chain(" finished"));
	
	return lsn;
	
    }
    catch ( Exception e )
    {
	Chain msg = e.getBaseMsg();

	_pDBMng->setTableSetRunState(tableSet, XML_OFFLINE_VALUE);    
	_pDBMng->setRecoveryMode(tabSetId, CegoDatabaseManager::OFF );	

	throw Exception(EXLOC, Chain("Recovery failed : ") + msg);
    }
}

long CegoRecoveryManager::dataFileRecovery(const Chain& tableSet, int tabSetId, long cplsn)
{

    bool endOfBackup = false;       
   
    long lsn = cplsn;

    while ( endOfBackup == false &&  _pDBMng->getRecoveryMode(tabSetId) == CegoDatabaseManager::ON )
    {
	
	Chain archLogFileName = _pDBMng->getArchiveLogName(tableSet, lsn);
	
	ListT<Chain> archIdList;
	ListT<Chain> archPathList;
	
	_pDBMng->getArchLogInfo(tabSetId, archIdList, archPathList);
	
	Chain *pArchLogPath = archPathList.First();

	Chain sourceFileName;
	
	bool fileFound = false;

	while ( pArchLogPath  && fileFound == false )
	{
			
	    // check if exists
	    
	    sourceFileName = *pArchLogPath + Chain("/") + archLogFileName;
	    
	    File checkFile(sourceFileName);

	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Checking file ") + sourceFileName);
	    
	    if ( checkFile.exists() )
	    {		    
		fileFound = true;		
	    }
	    else
	    {
		pArchLogPath = archPathList.Next();
	    }
	}

	if ( fileFound )
	{

	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Recovering offline logfile ") + sourceFileName + Chain(" ..."));

	    _pDBMng->setLogFile(tabSetId, sourceFileName, true);
	    _pDBMng->setLSN(tabSetId, lsn + 1);
	    
	    // recover offline logs
	    lsn = recoverCurrentDataFileLog(tabSetId, endOfBackup);
	    
	}
	else if ( _recoveryMode == LOCAL )
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Waiting for logfile ") + sourceFileName + Chain(" ..."));
	
	    Chain archRestoreProg = _pDBMng->getArchRestoreProg();
    
	    if ( archRestoreProg != Chain("NONE") )
	    {
		
		Chain *pArchLogPath = archPathList.First();
		Chain pathString;
		while ( pArchLogPath )
		{
		    pathString += *pArchLogPath;
		    pArchLogPath = archPathList.Next();
		    if ( pArchLogPath )
			pathString += Chain(":");
		}
				
		CommandExecuter cmdExe(_shellCmd);
		
		int archRestoreTimeout = _pDBMng->getArchRestoreTimeout();
		
		Chain restoreCmd = archRestoreProg + Chain(" ") + archLogFileName + Chain(" ") + pathString; 		
		_pDBMng->log(_modId, Logger::NOTICE, Chain("Triggering external log manager with <") + Chain(restoreCmd) + Chain(">"));
		int retCode = cmdExe.execute(restoreCmd, archRestoreTimeout);
		_pDBMng->log(_modId, Logger::NOTICE, Chain("External log manager returned : <") + Chain(retCode) + Chain(">"));
		if ( retCode == 0 )
		{
		    // file could be provided
		}
		else if ( retCode == 1 )
		{
		    endOfBackup = true;
		}
		else // ( retCode > 1 )
		{
		    Chain msg("External log maager failed");
		    throw Exception(EXLOC, msg);			    
		}
	    }
	    else
	    {
		Sleeper s;
		s.secSleep(LOGMNG_RECOVERY_DELAY);
	    }
	}
	else
	{
	    Sleeper s;
	    s.secSleep(LOGMNG_RECOVERY_DELAY);	    
	}
    }

    return lsn;
}

long CegoRecoveryManager::transactionRecovery(const Chain& tableSet, int tabSetId, long cplsn, int pit)
{
    bool endOfRecovery = false;    
    Chain sourceFileName;
    int ts = 0;

    long lsn = cplsn;

    RecoveryState rstate = CegoRecoveryManager::RECOK;
	    
    while ( _pDBMng->getRecoveryMode(tabSetId) == CegoDatabaseManager::ON 
	    && rstate != CegoRecoveryManager::RECPITREACHED
	    && endOfRecovery == false)
    {
	
	bool fileFound = false;   

	while ( fileFound == false && endOfRecovery == false &&  _pDBMng->getRecoveryMode(tabSetId) == CegoDatabaseManager::ON )
	{
	    
	    Chain archLogFileName = _pDBMng->getArchiveLogName(tableSet, lsn);
	    
	    ListT<Chain> archIdList;
	    ListT<Chain> archPathList;
	    
	    _pDBMng->getArchLogInfo(tabSetId, archIdList, archPathList);
	    
	    Chain *pArchLogPath = archPathList.First();
	    
	    while ( pArchLogPath  && fileFound == false)
	    {
		
		// check if exists
		
		sourceFileName = *pArchLogPath + Chain("/") + archLogFileName;
		
		File checkFile(sourceFileName);
		
		_pDBMng->log(_modId, Logger::NOTICE, Chain("Checking logfile ") + sourceFileName + Chain(" ..."));
		
		if ( checkFile.exists() )
		{		    
		    fileFound = true;		    
		}
		else
		{
		    pArchLogPath = archPathList.Next();
		}
	    }
	    
	    if ( fileFound == false && _recoveryMode == LOCAL)
	    {
		
		Chain archRestoreProg = _pDBMng->getArchRestoreProg();

		if ( archRestoreProg != Chain("NONE") )
		{	
		    
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Waiting for logfile ") + archLogFileName + Chain(" ..."));
		    
		    Chain *pArchLogPath = archPathList.First();
		    Chain pathString;
		    while ( pArchLogPath )
		    {
			pathString += *pArchLogPath;
			pArchLogPath = archPathList.Next();
			if ( pArchLogPath )
			    pathString += Chain(":");
		    }
		    
		    
		    CommandExecuter cmdExe(_shellCmd);
		    
		    int archRestoreTimeout = _pDBMng->getArchRestoreTimeout();
		    
		    Chain restoreCmd = archRestoreProg + Chain(" ") + archLogFileName + Chain(" ") + pathString; 
		    
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Triggering external log manager with <") + Chain(restoreCmd) + Chain(">"));
		    int retCode = cmdExe.execute(restoreCmd, archRestoreTimeout); 
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("External log manager returned : <") + Chain(retCode) + Chain(">"));
		    if ( retCode == 0 )
		    {
			// file could be provided
		    }
		    else if ( retCode == 1 )
		    {
			endOfRecovery = true;
		    }
		    else // ( retCode > 1 )
		    {
			Chain msg("External log manager failed");
			throw Exception(EXLOC, msg);			    
		    }		     
		}	      
		else
		{
		    endOfRecovery = true;
		}
	    }
	    if ( fileFound == false )
	    {
		Sleeper s;
		s.secSleep(LOGMNG_RECOVERY_DELAY);
	    }
	}

	// if we found the appropriate file, we try to recover it
	if ( fileFound )
	{

	    bool isRecovered = false;
	    while ( ! isRecovered  && _pDBMng->getRecoveryMode(tabSetId) == CegoDatabaseManager::ON )
	    {
		
		_pDBMng->log(_modId, Logger::NOTICE, Chain("Recovering offline logfile ") + sourceFileName + Chain(" ..."));
		_pDBMng->setLogFile(tabSetId, sourceFileName, true);
		
		_pDBMng->setLSN(tabSetId, lsn + 1);
		
		rstate =  recoverCurrentTransactionLog(tabSetId, pit, lsn, ts);
		
		if ( rstate == CegoRecoveryManager::RECOK )
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Transaction recovery ok"));
		    isRecovered=true;
		    lsn++;
		}
		else if ( rstate == CegoRecoveryManager::RECINCOMPLETE ) 
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Incomplete offline logfile ") + sourceFileName + Chain(" detected, waiting ..."));
		    Sleeper s;
		    s.secSleep(LOGMNG_RECOVERY_DELAY);
		}
		else if ( rstate == CegoRecoveryManager::RECPITREACHED )
		{	      
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("PIT reached"));
		    isRecovered=true;
		}
		else
		{
		    throw Exception(EXLOC, Chain("Unknwon recovery state"));
		}
	    }
	}


    }

    _pDBMng->log(_modId, Logger::NOTICE, Chain("Finished recovery loop ..."));

    // after recovery mode has been finished, we still have to check for the last logfile to recover
    if ( rstate == CegoRecoveryManager::RECINCOMPLETE )
    {
	while ( rstate != CegoRecoveryManager::RECOK )
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Recovering final offline logfile ") + sourceFileName + Chain(" ..."));

	    rstate =  recoverCurrentTransactionLog(tabSetId, pit, lsn, ts);	   

	    if ( rstate == CegoRecoveryManager::RECINCOMPLETE )
	    {
		_pDBMng->log(_modId, Logger::NOTICE, Chain("Waiting for final offline logfile ..."));
		Sleeper s;
		s.secSleep(LOGMNG_RECOVERY_DELAY);
	    }
	}		
    }

    _pGTM->writeCheckPoint(tableSet, false, false);

    return lsn;
}

int CegoRecoveryManager::recoverCurrentDataFileLog(int tabSetId, bool& endOfBackup)
{
    
    long lsn=0;

    _pDBMng->seekToStart(tabSetId);

    _pDBMng->log(_modId, Logger::NOTICE, Chain("Recovering logfile ... "));

    CegoLockHandler* pLockHandle = _pGTM->getLockHandle();

    // now recover to last lsn in log
    
    endOfBackup = false;
    CegoLogRecord logRec;
    while ( _pDBMng->logRead(tabSetId, logRec) && ! endOfBackup )
    {
	
	if ( logRec.getLSN() == _pDBMng->getLSN(tabSetId)  )
	{
	    
	    lsn=logRec.getLSN();
	    	    
	    switch ( logRec.getAction() ) 
	    {
		
	    case CegoLogRecord::LOGREC_BUPAGE:
		
#ifdef CGDEBUG	
		_pDBMng->log(_modId, Logger::DEBUG, Chain("Writing page (") + Chain(logRec.getFileId()) + Chain(",") + Chain(logRec.getPageId()) + Chain(")"));
#endif

		_pDBMng->writePage(logRec.getFileId(),logRec.getPageId(), 0, logRec.getData(), pLockHandle ); 
		break;
		
	    case CegoLogRecord::LOGREC_BUFBM:
	    {
#ifdef CGDEBUG	
		_pDBMng->log(_modId, Logger::DEBUG, Chain("Writing fbm for fileId ") + Chain(logRec.getFileId()));
#endif

		_pDBMng->writeFBM(logRec.getFileId(), (unsigned*)logRec.getData(), pLockHandle ); 
		break;
	    }
	    case CegoLogRecord::LOGREC_BUFIN:
	    {
		_pDBMng->log(_modId, Logger::NOTICE, Chain("Detected end of backup"));
		endOfBackup=true;
		break;
	    }
	    default:
		break;
	    }

	    _pDBMng->setLSN(tabSetId, lsn + 1);
	}
	else if ( logRec.getLSN() < _pDBMng->getLSN(tabSetId)  )
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Ignoring lsn ") + Chain(logRec.getLSN()) + Chain(" ( expected ") + Chain(_pDBMng->getLSN(tabSetId)) + Chain(")"));	    
	}
	else
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Skipping logfile lsn ") + Chain(logRec.getLSN()) + Chain(" ( expected ") + Chain(_pDBMng->getLSN(tabSetId)) + Chain(")"));	    
	    endOfBackup=true;

	}

	if ( logRec.getData() )
	    free ( logRec.getData() ) ;
	
    }
    
    _pDBMng->log(_modId, Logger::NOTICE, Chain("Datafile recovery finished with lsn ") + Chain(lsn));
   
    return lsn;
    
}

CegoRecoveryManager::RecoveryState CegoRecoveryManager::recoverCurrentTransactionLog(int tabSetId, int pit, long& lsn, int& ts)
{
    
    // lsn=0;

    // initially, we set the logfile to incomplete
    RecoveryState rstate = CegoRecoveryManager::RECINCOMPLETE;

    _pDBMng->seekToStart(tabSetId);

    _pDBMng->log(_modId, Logger::NOTICE, Chain("Recovering logfile ..."));

    // now recover to last lsn in log

    SetT<Chain> invalidateSet;
    CegoLogRecord logRec;
    while ( _pDBMng->logRead(tabSetId, logRec) && rstate != CegoRecoveryManager::RECPITREACHED )
    {
	// point-in-time recovery 
	if ( pit > 0 && logRec.getTS() > pit )
	{
	    Datetime d(logRec.getTS());
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Required point in time reached, Ignoring logentry  ") + Chain(logRec.getLSN()) 
			 + Chain(" ( ") + d.asChain() + Chain(" ) and higher"));
	    

	    rstate = CegoRecoveryManager::RECPITREACHED;
	    
	}
	else if ( logRec.getTS() <  ts )
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Old log detected, ignoring logentry ") + Chain(logRec.getLSN()) + Chain(" and higher"));

	    rstate = CegoRecoveryManager::RECPITREACHED;
	    
	}
	else
	{
	    if ( logRec.getLSN() == _pDBMng->getLSN(tabSetId)  )
	    {

		// we got a valid lsn, so the log file seems to be complete and we can switch from incomplete to ok
		rstate = CegoRecoveryManager::RECOK;

		lsn=logRec.getLSN();
		ts=logRec.getTS();
		int tid=logRec.getTID();

#ifdef CGDEBUG	
		_pDBMng->log(_modId, Logger::DEBUG, Chain("Recovering lsn ") + Chain(logRec.getLSN()) + Chain(" ..."));
#endif

		// here comes the recovery
		
		switch ( logRec.getAction() ) 
		{
		    
		case CegoLogRecord::LOGREC_CREATE:
		    
		    switch ( logRec.getObjType() )
		    {
			
		    case CegoObject::TABLE:
		    case CegoObject::SYSTEM:
		    {
			CegoTableObject to;
			to.decode(logRec.getData());
			_pGTM->createDataTable(tabSetId, logRec.getObjName(), logRec.getObjType(), to.getSchema());
			break;
		    }
		    case CegoObject::INDEX:
		    case CegoObject::PINDEX:
		    case CegoObject::UINDEX:
		    {
			CegoTableObject to;
			to.decode(logRec.getData());
			_pGTM->createIndexTable(tabSetId, logRec.getObjName(), to.getTabName(), to.getSchema(), logRec.getObjType());
			break;
		    }
		    case CegoObject::VIEW:
		    {
			CegoViewObject vo;
			vo.decode(logRec.getData());
			_pGTM->createViewObject(vo);
			break;		    
		    }
		    case CegoObject::PROCEDURE:
		    {
			CegoProcObject po;
			po.decode(logRec.getData());
			_pGTM->createProcObject(po);
			break;		    
		    }
		    case CegoObject::FKEY:
		    {
			CegoKeyObject ko;
			ko.decode(logRec.getData());
			_pGTM->createKeyObject(ko);
			break;
		    }
		    default:		    
			break;
		    }
		    break;
		case CegoLogRecord::LOGREC_DROP:
		{
		    _pGTM->removeObject(tabSetId, logRec.getObjName(), logRec.getObjType());
		    break;
		}
		case CegoLogRecord::LOGREC_ALTER:
		{
		    CegoTableObject aoe;
		    aoe.decode(logRec.getData());
		    _pGTM->alterTableObject(tabSetId, logRec.getObjName(), logRec.getObjType(), aoe);
		    break;
		}
		case CegoLogRecord::LOGREC_RENAME:
		{
		    Chain newObjName(logRec.getData(), logRec.getDataLen());
		    _pGTM->renameObject(tabSetId, logRec.getObjName(), logRec.getObjType(), newObjName );
		    break;   
		}
		case CegoLogRecord::LOGREC_BEGIN:
		{
		    // _pGTM->beginTransaction(tabSetId);
		    break;
		}
		case CegoLogRecord::LOGREC_COMMIT:
		{
		    _pGTM->setTID(tabSetId, tid);		    
		    _pGTM->commitTransaction(tabSetId);
		    break;
		}
		case CegoLogRecord::LOGREC_ABORT:
		{
		    _pGTM->setTID(tabSetId, tid);
		    _pGTM->rollbackTransaction(tabSetId);
		    break;
		}
		case CegoLogRecord::LOGREC_INSERT:
		{
		    
		    CegoTableObject oe;
		    _pGTM->getObject(tabSetId, logRec.getObjName(), CegoObject::TABLE, oe);

		    if ( invalidateSet.Find(logRec.getObjName()) == 0 )
		    {
			_pDBMng->log(_modId, Logger::NOTICE, Chain("Invalidating index for table ") + logRec.getObjName());		    
			_pGTM->invalidateIndexForTable(tabSetId, logRec.getObjName());
			invalidateSet.Insert(logRec.getObjName());
		    }
		    ListT<CegoField> fvl = oe.getSchema();
		    
		    int rectid = 0;
		    int tastep = 0;
		    CegoTupleState ts;
		    ListT<CegoBlob> blobList;
		    CegoQueryHelper qh;
		    qh.decodeFVL(fvl, blobList, logRec.getData(), logRec.getDataLen(), rectid, tastep, ts);
		    
		    CegoBlob *pBlob = blobList.First();
		    CegoField *pF = fvl.First();
		    while ( pBlob && pF )
		    {
	    	    
			int fileId;
			int pageId;
			
			_pGTM->putBlobData(tabSetId, pBlob->getBufPtr(), pBlob->getSize(), fileId, pageId);
			
			while ( pF->getValue().getType() != BLOB_TYPE && pF)
			    pF = fvl.Next();
			
			if ( pF )
			{
			    Chain blobRef = Chain("[") + Chain(fileId) + Chain(",") + Chain(pageId) + Chain("]");
			    CegoFieldValue fv(BLOB_TYPE, blobRef);
			    pF->setValue(fv);
			}
			else
			{
			    throw Exception(EXLOC, "Cannot get blob reference");
			}
			
			free ( pBlob->getBufPtr() );
			pBlob = blobList.Next();

		    }
		    
		    bool lastPage = true;
		    CegoDataPointer dp;

		    _pGTM->setTID(tabSetId, tid);
		    _pGTM->insertDataTable(oe, fvl, dp, lastPage);
		    
		    break;
		}
		case CegoLogRecord::LOGREC_DELETE:
		{
		  
		    CegoPredDesc *pPred = 0;	    
		    CegoQueryHelper qh;
		    Chain tableAlias;

		    qh.decodeDelRec(tableAlias, pPred, logRec.getData(), logRec.getDataLen(), _pGTM);

		    _delRecList.Insert(DeleteRecord(tid, tableAlias, pPred));
		    
		    break;
		}
		case CegoLogRecord::LOGREC_DELETE_COMMIT:
		{
 
		    DeleteRecord* delRecord = _delRecList.Find(DeleteRecord(tid));
		    
		    if ( delRecord )
		    {

			CegoTableObject oe;
			_pGTM->getObject(tabSetId, logRec.getObjName(), CegoObject::TABLE, oe);
			
			if ( invalidateSet.Find(logRec.getObjName()) == 0 )
			{
			    _pDBMng->log(_modId, Logger::NOTICE, Chain("Invalidating index for table ") + logRec.getObjName());		    
			    _pGTM->invalidateIndexForTable(tabSetId, logRec.getObjName());
			    invalidateSet.Insert(logRec.getObjName());
			}
			
			_pGTM->setTID(tabSetId, tid);
			_pGTM->deleteDataTable(oe, delRecord->getPred(), 0);		    

			delRecord->clean();
			_delRecList.Remove(DeleteRecord(tid));

			
		    }
		    else
		    {
			_pDBMng->log(_modId, Logger::NOTICE, Chain("Cannot find delete record for transaction ") + Chain(tid));
			
		    }
		    break;
		}
		case CegoLogRecord::LOGREC_UPDATE:
		{
		    
		    ListT<CegoField> updList;
		    ListT<CegoExpr*> exprList;

		    CegoPredDesc *pPred = 0;		    
		    CegoQueryHelper qh;
		    Chain tableAlias;
		    qh.decodeUpdRec(tableAlias, pPred, updList, exprList, logRec.getData(), logRec.getDataLen(), _pGTM);

		    _updRecList.Insert(UpdateRecord(tid, tableAlias, pPred, updList, exprList));

		    break;
		}
		case CegoLogRecord::LOGREC_UPDATE_COMMIT:
		{

		    UpdateRecord* updRecord = _updRecList.Find(UpdateRecord(tid));
		    
		    if ( updRecord )
		    {
		    
			if ( invalidateSet.Find(logRec.getObjName()) == 0 )
			{
			    _pDBMng->log(_modId, Logger::NOTICE, Chain("Invalidating index for table ") + logRec.getObjName());		    
			    _pGTM->invalidateIndexForTable(tabSetId, logRec.getObjName());
			    invalidateSet.Insert(logRec.getObjName());
			}

			_pGTM->setTID(tabSetId,tid); 
			_pGTM->updateDataTable(tabSetId, logRec.getObjName(),
					       updRecord->getTable(), 
					       updRecord->getPred(), 
					       updRecord->getUpdateList(), 
					       updRecord->getExprList(), 0);
			
			updRecord->clean();
			_updRecList.Remove(UpdateRecord(tid));
		    }
		    else
		    {
			_pDBMng->log(_modId, Logger::NOTICE, Chain("Cannot find update record for transaction ") + Chain(tid));			
		    }
		    break;
		}
		case CegoLogRecord::LOGREC_ADDCOUNTER:
		{
		    Chain counterName ( logRec.getData() );
		    long initValue = 0;
		    _pDBMng->addCounter(tabSetId, counterName, initValue);
		    break;
		}
		case CegoLogRecord::LOGREC_DELCOUNTER:
		{
		    Chain counterName ( logRec.getData() );
		    long initValue = 0;
		    _pDBMng->removeCounter(tabSetId, counterName);
		    break;
		}
		default:
		    break;
		}
		_pDBMng->setLSN(tabSetId, lsn + 1);
	    
	    }
	    else if ( logRec.getLSN() < _pDBMng->getLSN(tabSetId)  )
	    {
		_pDBMng->log(_modId, Logger::NOTICE, Chain("Ignoring lsn ") + Chain(logRec.getLSN()) 
			     + Chain(" ( expected ") + Chain(_pDBMng->getLSN(tabSetId)) + Chain(")"));

		rstate = CegoRecoveryManager::RECOK;

	    }
	    else
	    {
		_pDBMng->log(_modId, Logger::NOTICE, Chain("Skipping logfile lsn ") + Chain(logRec.getLSN()) 
			     + Chain(" ( expected ") + Chain(_pDBMng->getLSN(tabSetId)) + Chain(")"));


		rstate = CegoRecoveryManager::RECPITREACHED;
		
	    }
	}
		    
	if ( logRec.getData() )
	    free ( logRec.getData() ) ;

    }
    
    _pDBMng->log(_modId, Logger::NOTICE, Chain("Recovery finished with lsn ") + Chain(lsn));

    return rstate;
    
}


