///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoTransactionManager.cc
// -------------------------
// Cego transaction manager implementation
//     
// Design and Implementation by Bjoern Lemke
//     
// (C)opyright 2000-2013 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: CegoTransactionManager
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

#ifndef _REENTRANT
#define _REENTRANT    /* basic 3-lines for threads */
#endif

// base includes
#include <lfc/Exception.h>
#include <lfc/Sleeper.h>
#include <lfc/ThreadLock.h>
#include <lfc/Tokenizer.h>

// cego includes
#include "CegoTransactionManager.h"

// system includes
#include <string.h>
#include <stdlib.h>

#include "CegoDataType.h"
#include "CegoTupleState.h"
#include "CegoTableManager.h"
#include "CegoAVLIndexManager.h"

#define RBSEP "#"
#define SYS_RB "rbcatlog"
#define SYS_RBCOMMIT "rbcommit"
#define SYS_RBROLLBACK "rbrollback"

#define SYS_UPDTAB "updtab"

#define SYS_RB_TID "tid"
#define SYS_RB_FILEID "fileid"
#define SYS_RB_PAGEID "pageid"
#define SYS_RB_OFFSET "offset"
#define SYS_RB_TABLE "table"

#define SYS_UPDTAB_TID "tid"
#define SYS_UPDTAB_FILEID "fileid"
#define SYS_UPDTAB_PAGEID "pageid"
#define SYS_UPDTAB_OFFSET "offset"

extern bool __lockStatOn;

/////////////
// TAEntry //
/////////////

CegoTransactionManager::TAEntry::TAEntry()
{
}

CegoTransactionManager::TAEntry::TAEntry(const int tid)
{
    _tid = tid;
}

CegoTransactionManager::TAEntry::TAEntry(const int tid, const CegoBufferPage& bp, const CegoTableObject& oe)
{
    _tid = tid;
    _bp = bp;
    _oe = oe;
}

CegoTransactionManager::TAEntry::~TAEntry()
{
}

const int CegoTransactionManager::TAEntry::getTid() const
{
    return _tid;
}


CegoBufferPage& CegoTransactionManager::TAEntry::getBufferPage()
{
    return _bp;
}

CegoTableObject& CegoTransactionManager::TAEntry::getTableObject()
{
    return _oe;
}

CegoDataPointer CegoTransactionManager::TAEntry::getDataPointer()
{
    CegoDataPointer dp(_bp.getFileId(), _bp.getPageId(), _bp.getEntryPos());
    return dp;
}

CegoTransactionManager::TAEntry& CegoTransactionManager::TAEntry::operator = ( const CegoTransactionManager::TAEntry& t)
{
    _tid = t._tid;
    _bp = t._bp;
    _oe = t._oe;
    return (*this);
}

bool CegoTransactionManager::TAEntry::operator == ( const CegoTransactionManager::TAEntry& t)
{
    if ( _tid == t._tid )
	return true;
    return false;
}

CegoTransactionManager::CegoTransactionManager(CegoTableManager *pTM)
{
    _pTM = pTM;

    CegoFieldValue defVal;
    _rbcatSchema.Insert(CegoField(SYS_RB, SYS_RB, SYS_RB_TID, INT_TYPE, sizeof(int), defVal, true, 1));
    _rbcatSchema.Insert(CegoField(SYS_RB, SYS_RB, SYS_RB_FILEID, INT_TYPE, sizeof(int), defVal, true, 2));
    _rbcatSchema.Insert(CegoField(SYS_RB, SYS_RB, SYS_RB_PAGEID, INT_TYPE, sizeof(int), defVal, true, 3));
    _rbcatSchema.Insert(CegoField(SYS_RB, SYS_RB, SYS_RB_OFFSET, INT_TYPE, sizeof(int), defVal, true, 4));
    _rbcatSchema.Insert(CegoField(SYS_RB, SYS_RB, SYS_RB_TABLE, VARCHAR_TYPE, MAX_OBJNAME_LEN, defVal, false, 5));
    
    _updSchema.Insert(CegoField(SYS_UPDTAB, SYS_UPDTAB, SYS_UPDTAB_TID, INT_TYPE, sizeof(int), defVal, true, 1));
    _updSchema.Insert(CegoField(SYS_UPDTAB, SYS_UPDTAB, SYS_UPDTAB_FILEID, INT_TYPE, sizeof(int), defVal, true, 2));
    _updSchema.Insert(CegoField(SYS_UPDTAB, SYS_UPDTAB, SYS_UPDTAB_PAGEID, INT_TYPE, sizeof(int), defVal, true, 3));
    _updSchema.Insert(CegoField(SYS_UPDTAB, SYS_UPDTAB, SYS_UPDTAB_OFFSET, INT_TYPE, sizeof(int), defVal, true, 4));

    _pDBMng = pTM->getDBMng();

    _modId = _pDBMng->getModId("CegoTransactionManager");
}

CegoTransactionManager::~CegoTransactionManager()
{
    TAEntry *pTAE = _taList.First();
    while ( pTAE )
    {
	if ( pTAE->getBufferPage().isFixed() )
	    _pDBMng->bufferUnfix(pTAE->getBufferPage(), true, _pTM->getLockHandler());
	pTAE = _taList.Next();
    }   
}

void CegoTransactionManager::release(int tabSetId)
{
    TAEntry *pTAE = _taList.First();
    while ( pTAE )
    {
	_pDBMng->bufferUnfix(pTAE->getBufferPage(), true, _pTM->getLockHandler());
	pTAE = _taList.Next();
    }
}

void CegoTransactionManager::reorgSystemSpace(int tabSetId)
{
    // nothing to do
}

void CegoTransactionManager::initLock(int tabSetId)
{
    // nothing to do anymore
}

void CegoTransactionManager::newRBEntry(int tabSetId, int tid, int fileId, int pageId, int offset, const Chain& tableName)
{

    TAEntry *pTAE = _taList.Find( TAEntry(tid));

    CegoBufferPage bp;
    
    // if this is a new transaction, we have to create a rollback table for it
    if ( pTAE == 0 )
    {
	Chain taTable = Chain(SYS_RB) + Chain(RBSEP) + Chain(tid);
	
	CegoTableObject oe;

	// in case of crash recovery, the rb table might already exist
	if ( _pTM->objectExists(tabSetId, taTable, CegoObject::RBSEG) == false )
	{
	    _pTM->createDataTable(tabSetId, taTable, CegoObject::RBSEG, _rbcatSchema);
	}
	_pTM->getObjectWithFix(tabSetId, taTable, CegoObject::RBSEG, oe, bp);
	
	TAEntry tae(tid, bp, oe);

	_taList.Insert(tae);
	pTAE = _taList.Find( TAEntry(tid));
    }
    
    CegoFieldValue fv1(INT_TYPE, &tid, sizeof(int));
    CegoField f1(CegoField(SYS_RB, SYS_RB, SYS_RB_TID, INT_TYPE, sizeof(int), fv1, false, 1));
    
    CegoFieldValue fv2(INT_TYPE, &fileId, sizeof(int));
    CegoField f2(CegoField(SYS_RB, SYS_RB, SYS_RB_FILEID, INT_TYPE, sizeof(int), fv2, true, 2));	
    
    CegoFieldValue fv3(INT_TYPE, &pageId, sizeof(int));
    CegoField f3(CegoField(SYS_RB, SYS_RB, SYS_RB_PAGEID, INT_TYPE, sizeof(int), fv3, true, 3));	
    
    CegoFieldValue fv4(INT_TYPE, &offset, sizeof(int));
    CegoField f4(CegoField(SYS_RB, SYS_RB, SYS_RB_OFFSET, INT_TYPE, sizeof(int), fv4, true, 4));
    
    CegoFieldValue fv5(VARCHAR_TYPE, (char*)tableName, tableName.length());
    CegoField f5(CegoField(SYS_RB, SYS_RB, SYS_RB_TABLE, VARCHAR_TYPE, MAX_OBJNAME_LEN, fv5, false, 5));	
    
    ListT<CegoField> fl;
    fl.Insert(f1);
    fl.Insert(f2);
    fl.Insert(f3);
    fl.Insert(f4);
    fl.Insert(f5);
        
    int rbtid = 0;
    int rbtastep = 0;
    CegoTupleState ts = COMMITTED;
    
    char *pBufBase = 0;
    int buflen = 0;
    
    _qh.encodeFVL(rbtid, rbtastep, ts, fl, pBufBase, buflen);
    
    try
    {	
	CegoDataPointer dp = _pTM->insertData(pTAE->getDataPointer(), pTAE->getTableObject(), pBufBase, buflen, true);
    }
    catch ( Exception e )
    {
	free (pBufBase);
	throw e;
    }
    free (pBufBase);

    
}

/* 
   For the commit operation, we provide a method to give information about all affected table
   for this transaction. Setting a lock to all these tables before starting the commit avoids
   a possible deadlock
*/
   
void CegoTransactionManager::getTransactionAffectedTables(int tabSetId, int tid, SetT<Chain>& tableList)
{

    CegoObjectCursor *pOC = 0;

    try
    {
	
	TAEntry *pTAE = _taList.Find( TAEntry(tid) );

	if ( pTAE )
	{

	    Chain taTable = pTAE->getTableObject().getName();
	    pOC = _pTM->getObjectCursor(tabSetId, taTable, taTable, CegoObject::RBSEG);
	    
	    ListT<CegoField> schema = _rbcatSchema;
	    	    
	    CegoDataPointer dp;
	    bool moreTuple = _pTM->getFirstTuple(pOC, schema, dp);
	    
	    while (moreTuple)
	    {
		
		Chain tableName;
		CegoField *pF = schema.Find(CegoField(taTable, SYS_RB_TABLE));
		if (pF)
		{
		    tableName = Chain((char*)pF->getValue().getValue());			
		    tableList.Insert(tableName);			      
		}
		
		moreTuple = _pTM->getNextTuple(pOC, schema, dp);
		
	    }

	    pOC->abort();
	    delete pOC;
	    pOC = 0;
	    
	}	
	
    }
    catch ( Exception e )
    {
	if ( pOC )
	{
	    pOC->abort();
	    delete pOC;
	}
	throw e;
    }
}

long CegoTransactionManager::commitTransaction(int tabSetId, int tid)
{
    long opCount = 0;

    TAEntry *pTAE = _taList.Find( TAEntry(tid) );
    
    if ( pTAE == 0 )
    {	    
	
	// no transaction entry found. It seems, no 
	// modifying operations have been take place, so we can remove the segment and return immediately
	
	Chain taTable = Chain(SYS_RB) + Chain(RBSEP) + Chain(tid);
	if ( _pTM->objectExists(tabSetId, taTable, CegoObject::RBSEG) )
	{
	    _pTM->removeObject(tabSetId, taTable, CegoObject::RBSEG);
	}
	return opCount; 	
    }
    
    // We rename the rollback object to indicate the commit phase
    // In case of a crash recovery, the renamed objects have to be recovered
    Chain taTable = pTAE->getTableObject().getName();
    
    Chain commitTa = Chain(SYS_RBCOMMIT) + Chain(RBSEP) + Chain(tid);
    _pTM->renameObject(tabSetId, taTable, CegoObject::RBSEG, commitTa);
    pTAE->getTableObject().setName(commitTa);
    
    opCount = doCommit(tabSetId, commitTa);
    
    // Step 2 : Cleaning up rollback segment
    
    _pDBMng->bufferUnfix(pTAE->getBufferPage(), true, _pTM->getLockHandler());
    _pTM->removeObject(tabSetId, commitTa, CegoObject::RBSEG);
    _taList.Remove(TAEntry(tid));
    
    return opCount;
}

long CegoTransactionManager::doCommit(int tabSetId, const Chain& rbo)
{

    long opCount = 0;

    CegoObjectCursor *pOC = _pTM->getObjectCursor(tabSetId, rbo, rbo, CegoObject::RBSEG);
    
    ListT<CegoField> schema = _rbcatSchema;
    
    // Step 1 : Committing all attached tuples
    
    try  
    {
	
	CegoDataPointer rbdp;
	bool moreTuple = _pTM->getFirstTuple(pOC, schema, rbdp);
	
	Chain cachedTable;
	ListT<CegoField> cachedFvl;
	ListT<CegoTableObject> cachedIdxList;
	ListT<CegoBTreeObject> cachedBTreeList;
	ListT<CegoKeyObject> cachedKeyList;
	ListT<CegoCheckObject> cachedCheckList;
	
	while (moreTuple)
	{
	    
	    int fileId, pageId, offset;
	    
	    CegoField *pF = schema.Find(CegoField(SYS_RB, SYS_RB_FILEID));
	    if (pF)
		memcpy(&fileId, pF->getValue().getValue(), sizeof(int));
	    
	    pF = schema.Find(CegoField(SYS_RB, SYS_RB_PAGEID));
	    if (pF)
		memcpy(&pageId, pF->getValue().getValue(), sizeof(int));
	    
	    pF = schema.Find(CegoField(SYS_RB, SYS_RB_OFFSET));
	    if (pF)
		memcpy(&offset, pF->getValue().getValue(), sizeof(int));
	    
	    CegoDataPointer dp(fileId, pageId, offset);
	    
	    int tid;
	    int tastep;
	    CegoTupleState ts;
	    
	    _pTM->getTupleInfo(tabSetId, dp, tid, tastep, ts);			
	    _pTM->setTupleInfo(tabSetId, dp, 0, 0, COMMITTED);

	    opCount++;
	    if ( ts == DELETED || ts == OBSOLETE )
	    {
		
		Chain tableName;
		
		pF = schema.Find(CegoField(SYS_RB, SYS_RB_TABLE));
		if (pF)
		    tableName = Chain((char*)pF->getValue().getValue());
		
		
		if ( cachedTable != tableName )
		{
		    CegoTableObject oe;
		    _pTM->getObject(tabSetId, tableName, CegoObject::TABLE, oe);
		    cachedFvl = oe.getSchema();
		    cachedTable = tableName;
		    cachedIdxList.Empty();
		    cachedBTreeList.Empty();
		    cachedKeyList.Empty();
		    cachedCheckList.Empty();
		    _pTM->getObjectListByTable(tabSetId, cachedTable, cachedIdxList, cachedBTreeList, cachedKeyList, cachedCheckList);
		    
		}
		
		char* p;
		int len;
		unsigned long lockId;
		
		try
		{
		    lockId = _pTM->claimDataPtr(tabSetId, CegoLockHandler::READ, CegoBufferPool::SYNC, dp, p, len);
		    
		    int tid;
		    int tastep;
		    CegoTupleState ts;
		    _qh.decodeFVL(cachedFvl, p, len, tid, tastep, ts);
		    
		    _pTM->deleteDataTable(tabSetId, cachedTable, CegoObject::TABLE, dp, cachedFvl, cachedIdxList, cachedBTreeList, cachedKeyList, false);
		    
		}
		catch ( Exception e)
		{   
		    _pTM->releaseDataPtr(lockId);
		    throw Exception(EXLOC, Chain("Cannot commit transaction"), e );
		}
		_pTM->releaseDataPtr(lockId);		    
		
	    }

	    // set the rb datapointer to the current tid
	    // in case of a crash recovery, this tuple is ignored by the object cursor
	    // so we avoid a double commit

	    _pTM->setTupleInfo(tabSetId, rbdp, tid, 0, COMMITTED);
	    
	    moreTuple = _pTM->getNextTuple(pOC, schema, rbdp);
	}
    }
    catch ( Exception e )
    {
	pOC->abort();
	delete pOC;
	throw Exception(EXLOC, Chain("Cannot commit transaction"), e);
    }
    
    
    pOC->abort();
    delete pOC;

    return opCount;
}

void CegoTransactionManager::finishOpenTransaction(int tabSetId)
{
    
    _pDBMng->log(_modId, Logger::NOTICE, Chain("Finishing open transaction for tableset ") + Chain(tabSetId));
    ListT<Chain> rboList;    

    _pTM->getObjectList(tabSetId, CegoObject::RBSEG, rboList);
    
    Chain *pRBO = rboList.First();
    
    while ( pRBO )
    {
	
	_pDBMng->log(_modId, Logger::NOTICE, Chain("Treating ") + *pRBO);
	Tokenizer t(*pRBO, Chain(RBSEP));
	Chain taType;
	Chain taTid;

	t.nextToken(taType);
	t.nextToken(taTid);

	int tid = taTid.asInteger();
	
	if ( taType == Chain(SYS_RB) )
	{	    
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Rollback transaction ") + Chain(tid));
	    // transaction still was not in commit or rollback phase
	    // so we roll back
	    rollbackTransaction(tabSetId, tid);
	}
	else if ( taType == Chain(SYS_RBROLLBACK) )
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Finishing rollback for transaction ") + Chain(tid));
	    // transaction already was in rollback phase
	    // we have to finish rollback

	    doRollback(tabSetId, *pRBO);
	    _pTM->removeObject(tabSetId, *pRBO, CegoObject::RBSEG);
	}
	else if ( taType == Chain(SYS_RBCOMMIT) )
	{
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Finishing commit for transaction ") + Chain(tid));
	    // transaction already was in commit phase
	    // we have to finish commit
	    doCommit(tabSetId, *pRBO);
	    _pTM->removeObject(tabSetId, *pRBO, CegoObject::RBSEG);
	}

	pRBO = rboList.Next();
    }    
}

long CegoTransactionManager::rollbackTransaction(int tabSetId, int tid)
{
    
    long opCount = 0;

    TAEntry *pTAE = _taList.Find( TAEntry(tid) );
    
    if ( pTAE == 0 )
    {	    
	
	// no transaction entry found. It seems, no 
	// modifying operations have been take place, so we can remove the segemtn and return immediately
	
	Chain taTable = Chain(SYS_RB) + Chain(RBSEP) + Chain(tid);
	if ( _pTM->objectExists(tabSetId, taTable, CegoObject::RBSEG) )
	{
	    _pTM->removeObject(tabSetId, taTable, CegoObject::RBSEG);
	}
	return opCount; 
	
    }
    
    // rename the transaction object 
    Chain rbo = pTAE->getTableObject().getName();
    
    Chain rollbackTa = Chain(SYS_RBROLLBACK) + Chain(RBSEP) + Chain(tid);
    _pTM->renameObject(tabSetId, rbo, CegoObject::RBSEG, rollbackTa);
    pTAE->getTableObject().setName(rollbackTa);
    
    opCount = doRollback(tabSetId, rollbackTa);
    
    _pDBMng->bufferUnfix(pTAE->getBufferPage(), true, _pTM->getLockHandler());
    _pTM->removeObject(tabSetId, rollbackTa, CegoObject::RBSEG);	
    _taList.Remove(TAEntry(tid));
    
    return opCount;
    
}

long CegoTransactionManager::doRollback(int tabSetId, const Chain& rbo)
{

    long opCount = 0;

    CegoObjectCursor *pOC = 0;

    try
    {

	pOC = _pTM->getObjectCursor(tabSetId, rbo, rbo, CegoObject::RBSEG);
		
	ListT<CegoField> schema = _rbcatSchema;
	
	// step 1 ( tid setting )
	    
	CegoDataPointer rbdp;
	bool moreTuple = _pTM->getFirstTuple(pOC, schema, rbdp);
	
	Chain cachedTable;
	ListT<CegoField> cachedFvl;
	ListT<CegoTableObject> cachedIdxList;
	ListT<CegoBTreeObject> cachedBTreeList;
	ListT<CegoKeyObject> cachedKeyList;
	ListT<CegoCheckObject> cachedCheckList;
	
	while (moreTuple)
	{
	    
	    int fileId, pageId, offset;
	    Chain tableName;
	    
	    CegoField *pF = schema.Find(CegoField(SYS_RB, SYS_RB_FILEID));
	    if (pF)
		memcpy(&fileId, pF->getValue().getValue(), sizeof(int));
	    
	    pF = schema.Find(CegoField(SYS_RB, SYS_RB_PAGEID));
	    if (pF)
		memcpy(&pageId, pF->getValue().getValue(), sizeof(int));
	    
	    pF = schema.Find(CegoField(SYS_RB, SYS_RB_OFFSET));
	    if (pF)
		memcpy(&offset, pF->getValue().getValue(), sizeof(int));
	    
	    pF = schema.Find(CegoField(SYS_RB, SYS_RB_TABLE));
	    if (pF)
		tableName = Chain((char*)pF->getValue().getValue());
	    
	    
	    CegoDataPointer dp(fileId, pageId, offset);
	    
	    int tid;
	    int tastep;
	    CegoTupleState ts;
	    
	    _pTM->getTupleInfo(tabSetId, dp, tid, tastep, ts);			
	    _pTM->setTupleInfo(tabSetId, dp, 0, 0, COMMITTED);
	    opCount++;
	    
	    if ( ts == INSERTED || ts == OBSOLETE ) 
	    {
		
		if ( cachedTable != tableName )
		{
		    CegoTableObject oe;
		    _pTM->getObject(tabSetId, tableName, CegoObject::TABLE, oe);
		    cachedFvl = oe.getSchema();
		    cachedTable = tableName;
		    cachedIdxList.Empty();
		    cachedBTreeList.Empty();
		    cachedKeyList.Empty();
		    cachedCheckList.Empty();
		    _pTM->getObjectListByTable(tabSetId, cachedTable, cachedIdxList, cachedBTreeList, cachedKeyList, cachedCheckList);
		    
		}
		
		char* p;
		int len;
		unsigned long lockId;
		
		try
		{
		    lockId = _pTM->claimDataPtr(tabSetId, CegoLockHandler::READ, CegoBufferPool::SYNC, dp, p, len);
		    
		    int tid;
		    CegoTupleState ts;
		    _qh.decodeFVL(cachedFvl, p, len, tid, tastep, ts);
		    
		    _pTM->deleteDataTable(tabSetId, tableName, CegoObject::TABLE, dp, cachedFvl, cachedIdxList, cachedBTreeList, cachedKeyList, false, true);
		    
		}
		catch ( Exception e)
		{   
		    _pTM->releaseDataPtr(lockId);
		    
		    Chain msg;
		    e.pop(msg);
		    
		    // _pDBMng->log(_modId, Logger::NOTICE, Chain("Incomplete rollback for table ") + tableName + Chain(" : ") + msg);
		    throw Exception(EXLOC, Chain("Cannot delete entry for table"), e);
		}
		_pTM->releaseDataPtr(lockId);
		
	    }

	    // set the rb datapointer to the current tid
	    // in case of a crash recovery, this tuple is ignored by the object cursor
	    // so we avoid a double rollback

	    _pTM->setTupleInfo(tabSetId, rbdp, tid, 0, COMMITTED);

	    moreTuple = _pTM->getNextTuple(pOC, schema, rbdp);
	}
    }
    catch ( Exception e )
    {
	if ( pOC )
	{
	    pOC->abort();
	    delete pOC;
	}
	throw e;
    }
    
    pOC->abort();
    delete pOC;
	        
    return opCount;
}

void CegoTransactionManager::getTransactionInfo(int tabSetId, const Chain& rbo, int& numop)
{
    
    CegoObjectCursor *pOC;

    numop=0;

    if ( _pTM->objectExists(tabSetId, rbo, CegoObject::RBSEG) )
    {
	pOC = _pTM->getObjectCursor(tabSetId, rbo, rbo, CegoObject::RBSEG);

	try  {
	    
	    ListT<CegoField> schema = _rbcatSchema;
	    
	    CegoDataPointer dp;
	    bool moreTuple = _pTM->getFirstTuple(pOC, schema, dp);
	    
	    while (moreTuple)
	    {	       		
		numop++;
		moreTuple = _pTM->getNextTuple(pOC, schema, dp);
	    }
	}
	catch ( Exception e )
	{
	    pOC->abort();
	    delete pOC;
	    throw e;
	}
	
	pOC->abort();
	delete pOC;
    }
}


void CegoTransactionManager::recordUpdate(int tabSetId, int tid, const CegoDataPointer& dp )
{

    TAEntry *pUDE = _udList.Find( TAEntry(tid));
    
    if ( pUDE == 0 )
    {
	Chain udTable = Chain(SYS_UPDTAB) + Chain(tid);

	CegoBufferPage bp;	
	CegoTableObject oe;
	if ( _pTM->objectExists(tabSetId, udTable, CegoObject::RBSEG) == false )
	{
	    _pTM->createDataTable(tabSetId, udTable, CegoObject::RBSEG, _updSchema);
	}
	_pTM->getObjectWithFix(tabSetId, udTable, CegoObject::RBSEG, oe, bp);

	TAEntry ude(tid, bp, oe);

	_udList.Insert(ude);
	pUDE = _udList.Find( TAEntry(tid));
    }

    int fileId = dp.getFileId();
    int pageId = dp.getPageId();
    int offset = dp.getOffset();

    CegoFieldValue fv1(INT_TYPE, &tid, sizeof(int));
    CegoField f1(CegoField(SYS_UPDTAB, SYS_UPDTAB, SYS_UPDTAB_TID, INT_TYPE, sizeof(int), fv1, true, 1));	

    CegoFieldValue fv2(INT_TYPE, &fileId, sizeof(int));
    CegoField f2(CegoField(SYS_UPDTAB, SYS_UPDTAB, SYS_UPDTAB_FILEID, INT_TYPE, sizeof(int), fv2, true, 2));	

    CegoFieldValue fv3(INT_TYPE, &pageId, sizeof(int));
    CegoField f3(CegoField(SYS_UPDTAB, SYS_UPDTAB, SYS_UPDTAB_PAGEID, INT_TYPE, sizeof(int), fv3, true, 3));	

    CegoFieldValue fv4(INT_TYPE, &offset, sizeof(int));
    CegoField f4(CegoField(SYS_UPDTAB, SYS_UPDTAB, SYS_UPDTAB_OFFSET, INT_TYPE, sizeof(int), fv4, true, 4));

    ListT<CegoField> fl;
    fl.Insert(f1);
    fl.Insert(f2);
    fl.Insert(f3);
    fl.Insert(f4);

    int rbtid = 0;
    int rbtastep = 0;
    CegoTupleState ts = COMMITTED;

    char *pBufBase = 0;
    int buflen = 0;

    try
    {    
	_qh.encodeFVL(rbtid, rbtastep, ts, fl, pBufBase, buflen);    	
	CegoDataPointer dp = _pTM->insertData(pUDE->getDataPointer(), pUDE->getTableObject(), pBufBase, buflen, true);
    }
    catch ( Exception e )
    {
	if ( pBufBase )
	    free (pBufBase);
	
	throw e;
    }
    
    free (pBufBase);
    
}

/* 
   the commitUpdate method completes a transaction oriented update operation, where a index 
   was used to find matching tuples.
   In this case, the ( virgin ) index still has to expand for the updated tuples
*/

void CegoTransactionManager::commitUpdate(int tabSetId, const Chain& idxName, int tid, bool doAppend)
{

    CegoObjectCursor *pOC = 0;
    unsigned long lockId = 0;
    CegoBufferPage bp;
    bool isFixed = false;
    
    try
    {

	TAEntry *pUDE = _udList.Find( TAEntry(tid) );
	
	if ( pUDE == 0 )
	{	    
	    // no update entry found. It seems, no 
	    // modifying operations have been take place, so we can retrn immediately
	    return;
	    // throw Exception(EXLOC, Chain("Cannot find transaction ") + Chain(tid));
	}


	Chain udTable = pUDE->getTableObject().getName();
	
	CegoTableObject ioe;
	_pTM->getObjectWithFix(tabSetId, idxName, CegoObject::INDEX, ioe, bp);			
	isFixed = true;
	
	pOC = _pTM->getObjectCursor(tabSetId, udTable, udTable, CegoObject::RBSEG);
	

	ListT<CegoField> schema = _updSchema;
		
	CegoDataPointer updp;
	bool moreTuple = _pTM->getFirstTuple(pOC, schema, updp);
    
	while ( moreTuple)
	{
	    
	    int fileId, pageId, offset;
	    
	    CegoField *pF = schema.Find(CegoField(SYS_UPDTAB, SYS_UPDTAB_FILEID));
	    if (pF)
		fileId = *(int*)pF->getValue().getValue();
	    
	    pF = schema.Find(CegoField(SYS_UPDTAB, SYS_UPDTAB_PAGEID));
	    if (pF)
		pageId = *(int*)pF->getValue().getValue();
	    
	    pF = schema.Find(CegoField(SYS_UPDTAB, SYS_UPDTAB_OFFSET));
	    if (pF)
		offset = *(int*)pF->getValue().getValue();
	    
	    CegoDataPointer dp(fileId, pageId, offset);
	    
	    ListT<CegoField> fvl = ioe.getSchema();
	    
	    char* p;
	    int len;
	    
	    CegoDataPointer sysEntry(bp.getFileId(), bp.getPageId(), bp.getEntryPos());
	
	    
	    lockId = _pTM->claimDataPtr(tabSetId, CegoLockHandler::READ, CegoBufferPool::SYNC, dp, p, len);
	    
	    int tid;
	    int tastep;
	    CegoTupleState ts;
	    _qh.decodeFVL(fvl, p, len, tid, tastep, ts);
	    
	    char ip[TABMNG_MAXINDEXVALUE];
	    int ilen = TABMNG_MAXINDEXVALUE;
	    int idxLen;
	    CegoDataPointer ritp;
	    	   
	    _pTM->extractIndexValue(fvl, ioe.getSchema(), ip, ilen, idxLen);
	
	    CegoAVLIndexManager idxMng(_pTM);
	    idxMng.insertNativeIndexTable(ioe, sysEntry, dp, ip, idxLen, tid, doAppend, ritp);

	    _pTM->releaseDataPtr(lockId);
	    lockId = 0;
	    

	    moreTuple = _pTM->getNextTuple(pOC, schema, updp);
	    
	}
	
	delete pOC;
	pOC = 0;
	
	// now we can release the index root page
	_pTM->getDBMng()->bufferUnfix(bp, true, _pTM->getLockHandler());
	isFixed = false;

	_pTM->removeObject(tabSetId, udTable, CegoObject::RBSEG);

	// release the update commit object
	_pTM->getDBMng()->bufferUnfix(pUDE->getBufferPage(), true, _pTM->getLockHandler());	
	_udList.Remove(TAEntry(tid));

    }
    catch ( Exception e )
    {
	if ( lockId )
	{
	    _pTM->releaseDataPtr(lockId);
	}
	if ( isFixed )
	{
	    _pTM->getDBMng()->bufferUnfix(bp, true, _pTM->getLockHandler());
	}
	if ( pOC )
	{
	    pOC->abort();
	    delete pOC;
	}
	throw e;
    }

    if ( pOC )
    {
	pOC->abort();
	delete pOC;
    }
}

