///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoDistCursor.cc
// -----------------
// Cego global table cursor 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: CegoDistCursor
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

#include <lfc/Net.h>

#include "CegoDistCursor.h"
#include "CegoDistManager.h"
#include "CegoSelect.h"
#include "CegoTableObject.h"
#include "CegoJoinObject.h"
#include "CegoXMLdef.h"


CegoDistCursor::CegoDistCursor()
{
}

CegoDistCursor::CegoDistCursor(CegoDistManager* pGTM, CegoContentObject *pCO)
{

    _isLocal = true;
    _moreTuple = false;
    
    _pSelect = 0;
    _pSH = 0;
    _pGTM = pGTM;
    _tableName = pCO->getTabName();
    
    _tableAlias = pCO->getName();
    _tabSetId = pCO->getTabSetId();

    _pDBMng = _pGTM->getDBMng(); 
    
    _pCO = pCO;

    _pTCLeft = 0;
    _pTCRight = 0;
    _pTC = 0;
    _pC = 0;

    _idxMatch = CegoAttrCond::INAPP;
    
    _modId = _pDBMng->getModId("CegoDistCursor");


}

CegoDistCursor::~CegoDistCursor()
{
    if ( _pTC )
    {
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Releasing global table cursor for ") + _tableName);
#endif

	try {
	    
	    _pDBMng->unuseObject(_tabSetId, _tableName, CegoObject::TABLE, CegoDatabaseManager::SHARED);
	}
	catch ( Exception e )
	{
	    Chain msg;
	    e.pop(msg);
	    _pDBMng->log(_modId, Logger::LOGERR, Chain("Ignoring error : ") + msg);
	}
	delete _pTC;
    }
    if ( _pSelect )
    {
	try
	{
	    _pSelect->cleanUp();
	    _pDBMng->unuseObject(_tabSetId, _tableName, CegoObject::VIEW, CegoDatabaseManager::SHARED);
	}
	catch ( Exception e )
	{
	    Chain msg;
	    e.pop(msg);
	    _pDBMng->log(_modId, Logger::LOGERR, Chain("Ignoring error : ") + msg);
	}
    }
    if ( _pCO->getType() == CegoObject::JOIN )
    {

	CegoJoinObject *pJCO = (CegoJoinObject*)_pCO;

	CegoContentObject *pCOLeft = pJCO->getLeftObject();
	CegoContentObject *pCORight = pJCO->getRightObject();

#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Releasing global table cursor for ") + pCOLeft->getTabName());
#endif
	
	if ( pCOLeft->getType() == CegoObject::VIEW || pCOLeft->getType() == CegoObject::TABLE )
	    _pDBMng->unuseObject(pCOLeft->getTabSetId(), pCOLeft->getTabName(), pCOLeft->getType(), CegoDatabaseManager::SHARED);
	delete _pTCLeft;
	
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Releasing global table cursor for ") + pCORight->getTabName());
#endif
	
	if ( pCORight->getType() == CegoObject::VIEW || pCORight->getType() == CegoObject::TABLE )
	    _pDBMng->unuseObject(pCORight->getTabSetId(), pCORight->getTabName(), pCORight->getType(), CegoDatabaseManager::SHARED);
	delete _pTCRight;
		
    }

    if ( _pC )
    {
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Releasing object cursor for ") + _tableName);
#endif
	delete _pC;
    }

    if ( _pSH )
    {
	_pDBMng->releaseSession(_pSH);
    }
}

void CegoDistCursor::checkType()
{

    if ( _pCO->getType() == CegoObject::VIEW )

    {
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("View ") + _tableName + Chain(" detected as the cursor source"));
#endif

	_pDBMng->useObject(_tabSetId, _tableName, CegoObject::VIEW, CegoDatabaseManager::SHARED, _pGTM->getThreadId());

	CegoView *pView;

	try
	{
	    pView = _pGTM->getView(_tabSetId, _tableName);
	}
	catch ( Exception e )
	{
	    _pDBMng->unuseObject(_tabSetId, _tableName, CegoObject::VIEW, CegoDatabaseManager::SHARED);
	    throw e;
	}
	_pSelect = pView->getSelect();
	_pSelect->cleanUp();
	
    }
    else if ( _pCO->getType() == CegoObject::TABLE )
    {

#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Table ") + _tableName + Chain(" detected as the cursor source"));
#endif

	Chain tableSet = _pDBMng->getTabSetName(_tabSetId);

	CegoTableObject oe;
	_pGTM->getDistObject(tableSet, _tableName, CegoObject::TABLE, oe);
	
	if ( oe.isLocal() )
	{

#ifdef CGDEBUG
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Native Table ") + _tableName + Chain(" detected as the cursor source"));
#endif

	    _pTC = new CegoTableCursor(_pGTM, _tabSetId, _tableName, false);	
	    _pDBMng->useObject(_tabSetId, _tableName, CegoObject::TABLE, CegoDatabaseManager::SHARED, _pGTM->getThreadId());
	    _isLocal = true;
	}
	else
	{

#ifdef CGDEBUG
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Remote Table ") + _tableName + Chain(" detected as the cursor source"));
#endif
	    _isLocal = false;

	    int tabSetId = oe.getTabSetId();
	    Chain hostName = _pDBMng->getPrimary(tabSetId);
	    int portNo;
	    _pDBMng->getDataPort(portNo);

	    _schema = oe.getSchema();

	    Chain user;
	    Chain password;
	    
	    _pGTM->getActiveUser(tableSet, user, password);
	    
	    _pSH = _pDBMng->allocateSession(hostName, tableSet, user, password);
	    	    
	    _pSH->reqTableDataOp(tabSetId, _tableName,  CegoObject::TABLE);

	}      
    }
    else if ( _pCO->getType() == CegoObject::JOIN )
    {	

	CegoJoinObject *pJCO = (CegoJoinObject*)_pCO;

	CegoContentObject *pCOLeft = pJCO->getLeftObject();
	CegoContentObject *pCORight = pJCO->getRightObject();
	
	_pTCLeft = new CegoDistCursor(_pGTM, pCOLeft);
	_pTCLeft->checkType();

	if ( pCOLeft->getType() == CegoObject::VIEW || pCOLeft->getType() == CegoObject::TABLE )
	    _pDBMng->useObject(pCOLeft->getTabSetId(), pCOLeft->getTabName(), pCOLeft->getType(), CegoDatabaseManager::SHARED, _pGTM->getThreadId());
	
	_pTCRight = new CegoDistCursor(_pGTM, pCORight);
	_pTCRight->checkType();

	if ( pCORight->getType() == CegoObject::VIEW || pCORight->getType() == CegoObject::TABLE )
	    _pDBMng->useObject(pCORight->getTabSetId(), pCORight->getTabName(), pCORight->getType(), CegoDatabaseManager::SHARED, _pGTM->getThreadId());
	
    }

}

void CegoDistCursor::distSetup()
{

    _isFirst=true;
    _doEval = false;
    _idxMatch = CegoAttrCond::INAPP;
    _isAttrCondValid = false;

    if ( _pCO->getType() == CegoObject::VIEW )
    {
	_pSelect->setTabSetId(_tabSetId);
	CegoAttrCond noCond;
	_pSelect->setViewCond(noCond);
	_pSelect->prepare();
	_pSelect->setParentJoinBuf();

	_pSelect->checkValidRef();
    }
    else if ( _pCO->getType() == CegoObject::TABLE )
    {
	// nothing to do
    }
    else if ( _pCO->getType() == CegoObject::SYSTEM )
    {
	sysSetup();
    }
    else if ( _pCO->getType() == CegoObject::JOIN )
    {
	CegoAttrCond attrCond;
	joinSetup(attrCond);
    }
}

void CegoDistCursor::distSetup( const CegoAttrCond& attrCond)
{

    // cout << "Attr Comp Size = " << attrCond.numComp() << endl;

    if ( attrCond.numComp() == 0 )
	return distSetup();

    _isAttrCondValid = false;
    _idxMatch = CegoAttrCond::INAPP;
    _isFirst=true;
    _doEval = false;
    
    if ( _pCO->getType() == CegoObject::VIEW )
    {	
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Setting view condition ") + attrCond.toChain());
#endif
	_pSelect->setViewCond(attrCond);
	_pSelect->setTabSetId(_tabSetId);
	_pSelect->prepare();
	_pSelect->setParentJoinBuf();

	_pSelect->checkValidRef();

	_cursorCond = attrCond;

	_doEval=false;
    }
    else if ( _pCO->getType() == CegoObject::TABLE )
    {
	if ( _isLocal == true )
	{
	    _idxMatch = _pTC->setup(attrCond);
	    
	    if ( _idxMatch != CegoAttrCond::FULL )
	    {	       
		_doEval=true;
	    }

	    _cursorCond = attrCond;

	}
    }
    else if ( _pCO->getType() == CegoObject::SYSTEM )
    {
	sysSetup();
	_cursorCond = attrCond;
	_doEval = true;	
    }
    else if ( _pCO->getType() == CegoObject::JOIN )
    {
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Setting join condition ") + attrCond.toChain());	
#endif

	joinSetup(attrCond);
    }
}

bool CegoDistCursor::nextTuple(ListT<CegoField>* flArray, int pos, int size)
{
    while ( getTuple(flArray, pos, size) ) 
    {
	if ( _doEval == true )
	{
	    if ( evalCondition(flArray, pos, size) )
	    {
		return true;
	    }
	}
	else
	{
	    return true;
	}
    }

    return false;
}

bool CegoDistCursor::nextTuple(ListT<CegoField>& fl)
{
    return nextTuple(&fl, 0, 1);
}

bool CegoDistCursor::evalCondition(ListT<CegoField>* flArray, int pos, int size)
{

    CegoAttrComp *pAC = _cursorCond.getAttrCompSet().First();

    while ( pAC )
    {

	CegoField *pF = 0;
	
	int i=0;
	int lpos = pos;
	while ( pF == 0 && i < size)
	{	   
	    if ( _pCO->getType() == CegoObject::VIEW || _pCO->getType() == CegoObject::TABLE )
		pF = flArray[pos].Find( CegoField(_tableAlias, pAC->getAttrName()));
	    else
		pF = flArray[pos].Find( CegoField(Chain(), pAC->getAttrName()));
	    lpos++;
	    i++;
	}
	
	if ( pF )
	{
	    if ( pAC->getCompMode() == CegoAttrComp::BTWN )
	    {
		if ( ( (CegoFieldValue)pF->getValue() >= (CegoFieldValue)(pAC->getFieldValue()) 
		       && (CegoFieldValue)pF->getValue() <= (CegoFieldValue)(pAC->getFieldValue2()) ) == false )
		    
		    return false;	    
	    }
	    else if ( pAC->getCompMode() == CegoAttrComp::ISLIKE )
	    {		
		if ( pAC->getMatcher()->match(pF->getValue().valAsChain()) == false)		
		    return false;
	       
	    }
	    else if ( pAC->getCompMode() == CegoAttrComp::ISNOTLIKE )
	    {
		if ( pAC->getMatcher()->match(pF->getValue().valAsChain()))
		    return false;		
	    }
	    else
	    {
		switch ( pAC->getComparison() )
		{	    
		case EQUAL:
		{
		    if ( ( (CegoFieldValue)pF->getValue() == (CegoFieldValue)(pAC->getFieldValue()) ) == false )		    
			return false;		    
		    break;
		}
		case NOT_EQUAL:
		{
		    if ( ( (CegoFieldValue)pF->getValue() != (CegoFieldValue)(pAC->getFieldValue()) ) == false )		    
			return false;	    
		    break;
		}
		case LESS_THAN:
		{
		    if ( ( (CegoFieldValue)pF->getValue() < (CegoFieldValue)(pAC->getFieldValue()) ) == false )		     
			return false;
		    break;
		}
		case MORE_THAN: 
		{
		    if ( ( (CegoFieldValue)pF->getValue() > (CegoFieldValue)(pAC->getFieldValue()) ) == false )
			return false;
		    break;
		}
		case LESS_EQUAL_THAN:
		{
		    if ( ( (CegoFieldValue)pF->getValue() <= (CegoFieldValue)(pAC->getFieldValue()) ) == false )
			return false;
		    break;
		}
		case MORE_EQUAL_THAN:
		{
		    if ( ( (CegoFieldValue)pF->getValue() >= (CegoFieldValue)(pAC->getFieldValue()) ) == false )
			return false;	    
		    break;
		}
		}
	    }
	}
	else
	{
	    if ( pAC->isSetup() == false )
	    {
		Chain msg = Chain("Cannot get value for attribute ") + pAC->getAttrName();
		throw Exception(EXLOC, msg);
	    }
	}
	pAC = _cursorCond.getAttrCompSet().Next();
    }
    return true;
}

bool CegoDistCursor::getTuple(ListT<CegoField>* flArray, int offset, int size)
{

    if ( _pCO->getType() == CegoObject::VIEW )
    {
	try 
	{

	    _moreTuple = _pSelect->nextTuple(flArray[offset]);

	    if ( _moreTuple )
	    {

		CegoField *pF = flArray[offset].First();
		while ( pF ) 
		{
		    pF->setTableName(_tableName);
		    pF->setTableAlias(_tableAlias);
		    pF = flArray[offset].Next();
		}
	    }
	    return _moreTuple;
	}
	catch ( Exception e )
	{
	    _pSelect->cleanUp();
	    throw e;
	}
    }
    else if ( _pCO->getType() == CegoObject::TABLE )
    {
	if ( _isLocal )
	{
	    
	    CegoDataPointer dp;
	    if ( _isFirst )
	    {
		_moreTuple = _pTC->getFirst(flArray[offset], dp);
		_isFirst = false;
	    }
	    else
		_moreTuple = _pTC->getNext(flArray[offset], dp);	    
	    return _moreTuple;
	    
	}
	else
	{
	    flArray[offset] = _schema;
	    _moreTuple = false;

	    if ( _pSH->receiveTableData(flArray[offset]) == CegoDbHandler::DB_DATA )
	    {
		_moreTuple = true;
	    }

	    return _moreTuple;
	    
	}
    }
    else if ( _pCO->getType() == CegoObject::SYSTEM )
    {

	if ( _pC )
	{
	    
	    CegoDataPointer dp;

	    int len;
	    char *pc;
	    if ( _isFirst )
	    {	       
		pc = (char*)_pC->getFirst(len, dp);		
		_isFirst=false;
	    }
	    else
	    {
		pc = (char*)_pC->getNext(len, dp);
	    }

	    if ( pc && len > 0 )
	    {
		int tid;
		int tastep;	    
		CegoTupleState ts;
		
		CegoQueryHelper qh;
		qh.decodeFVL(flArray[offset], pc, len, tid, tastep, ts);			    
		return true;   
	    }
	    else
	    {
		return false;
	    }
	}
	else
	{
	
	    Chain *pName = 0;
	    if ( _isFirst )
	    {
		_isFirst = false;
		pName = _sysObjList.First();
	    }	
	    else
	    {
		pName = _sysObjList.Next();	    
	    }
	    
	    if ( pName )
	    {
		CegoField *pF = flArray[offset].Find(CegoField(_tableAlias, Chain(SYSTAB_NAME_ATTR)));
		if ( pF )		    
		{
		    pF->setValue( CegoFieldValue(VARCHAR_TYPE, *pName));
		}
		
		if ( _tableName == Chain(SYSTAB_TABLE_ID) 
		     || _tableName == Chain(SYSTAB_INDEX_ID)
		     || _tableName == Chain(SYSTAB_BTREE_ID) )
		{
		    
		    int pageCount;
		    
		    if ( _tableName == Chain(SYSTAB_TABLE_ID)  )
			pageCount = _pGTM->getPageCount(_tabSetId, *pName, CegoObject::TABLE);
		    else if ( _tableName == Chain(SYSTAB_INDEX_ID)  )
			pageCount = _pGTM->getPageCount(_tabSetId, *pName, CegoObject::INDEX);
		    else if ( _tableName == Chain(SYSTAB_BTREE_ID)  )
			pageCount = _pGTM->getPageCount(_tabSetId, *pName, CegoObject::BTREE);

		    CegoField *pF = flArray[offset].Find(CegoField(_tableAlias, Chain(SYSTAB_SIZE_ATTR)));
		    if ( pF )		    
		    {
			pF->setValue( CegoFieldValue(INT_TYPE, pageCount));
		    }
		    pF = flArray[offset].Find(CegoField(_tableAlias, Chain(SYSTAB_STATUS_ATTR)));
		    if ( pF )
		    {
			if ( pageCount > 0 )
			    pF->setValue( CegoFieldValue(VARCHAR_TYPE, Chain("valid")));
			else
			    pF->setValue( CegoFieldValue(VARCHAR_TYPE, Chain("invalid")));
		    }
		    
		}
		else if ( _tableName == Chain(SYSTAB_VIEW_ID) )
		{
		    Chain status("not compiled");
		    if ( _pGTM->checkCompView(_tabSetId, *pName) )
			status = Chain("compiled");
		    
		    CegoField *pF = flArray[offset].Find(CegoField(_tableAlias, Chain(SYSTAB_STATUS_ATTR)));
		    if ( pF )		    
		    {
			pF->setValue( CegoFieldValue(VARCHAR_TYPE, status));
		    }		
		}
		else if ( _tableName == Chain(SYSTAB_PROC_ID) )
		{
		    Chain status("not compiled");
		    if ( _pGTM->checkCompProcedure(_tabSetId, *pName) )
			status = Chain("compiled");
		    
		    CegoField *pF = flArray[offset].Find(CegoField(_tableAlias, Chain(SYSTAB_STATUS_ATTR)));
		    if ( pF )		    
		    {
			pF->setValue( CegoFieldValue(VARCHAR_TYPE, status));
		    }				
		}
		_moreTuple = true;	    
	    }
	    else
	    {
		_moreTuple = false;
	    }
	}

	return _moreTuple;
    }
    else if ( _pCO->getType() == CegoObject::JOIN )
    {

	if ( _isFirst )
	{
	    _moreLeft = true;
	    _moreRight = true;
	}

	CegoJoinObject *pJO = (CegoJoinObject*)_pCO;
	
	if ( pJO->getJoinType() == CegoJoinObject::INNER )
	{
	    bool isEvaluated = false;
	    
	    while ( isEvaluated == false && _moreLeft )
	    {
		if ( _isFirst )
		{		
		    _pTCLeft->distSetup(_cursorCond);
		    
		    _moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
		    
		    if ( _moreLeft )
		    {			
			_innerCond.setup(flArray, offset, size-1);

			_pTCRight->reset();

			if ( _isAttrCondValid )
			    _pTCRight->distSetup(_innerCond);
			else
			    _pTCRight->distSetup();

			_moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
		    }

		    /*
		    if ( _moreLeft )
			cout << "MoreLeft true" << endl;
		    if ( _moreRight )
			cout << "MoreRight true" << endl;
			

		    cout << "First tuple finished.." << endl;
		    */

		    _isFirst = false;
		}
		else if ( _moreRight )
		{

		    _moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
		    if ( _moreRight == false )
		    {
			_moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
			if ( _moreLeft )
			{
			    _innerCond.setup(flArray, offset, size-1);

			    _pTCRight->reset();

			    if ( _isAttrCondValid )
				_pTCRight->distSetup(_innerCond);
			    else
				_pTCRight->distSetup();

			    _moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
			}
		    }
		}
		else if ( _moreLeft )
		{
		    _moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
		    if ( _moreLeft )
		    {
			_innerCond.setup(flArray, offset, size-1);

			_pTCRight->reset();

			if ( _isAttrCondValid )
			    _pTCRight->distSetup(_innerCond);
			else
			    _pTCRight->distSetup();

			_moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
		    }
		}

		if ( _moreLeft && _moreRight )
		{
		    if ( _evalPredicate )
		    {
			
			CegoQueryHelper queryHelper;
			
			pJO->getPredDesc()->clearAttrCache();
			
			isEvaluated = queryHelper.evalPredicate(0,
								0,
								0,
								flArray,
								offset,
								size,
								pJO->getPredDesc(),
								0);
		    }
		    else
		    {			
			isEvaluated = true;
		    }
		}
	    }	    
	    
	    return _moreLeft && _moreRight;
	}
	else if ( pJO->getJoinType() == CegoJoinObject::LEFTOUTER )
	{

	    bool isEvaluated = false;
	    
	    while ( isEvaluated == false )
	    {
		
		if ( _isFirst )
		{
		    _pTCLeft->distSetup(_cursorCond);
		    
		    _moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
		    
		    if ( _moreLeft )
		    {
			_innerCond.setup(flArray, offset, size-1);
			
			_pTCRight->reset();
			
			if ( _isAttrCondValid )
			{
			    _pTCRight->distSetup(_innerCond);
			}
			else
			{
			    _pTCRight->distSetup();
			}
			
			_moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
		    }
		    _isFirst = false;
		}
		else if ( _moreRight )
		{
		    _moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
		    
		    if ( _moreRight == false )
		    {
			_moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
			if ( _moreLeft )
			{
			    _innerCond.setup(flArray, offset, size-1);
			    
			    _pTCRight->reset();
			    
			    if ( _isAttrCondValid )
			    {
				_pTCRight->distSetup(_innerCond);
			    }
			    else
			    {
				_pTCRight->distSetup();
			    }
			    
			    _moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
			}
		    }		
		}
		else if ( _moreLeft )
		{
		    _moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
		    if ( _moreLeft )
		    {
			_innerCond.setup(flArray, offset, size-1);
			
			_pTCRight->reset();
			
			if ( _isAttrCondValid )
			{
			    _pTCRight->distSetup(_innerCond);
			}
			else
			{
			    _pTCRight->distSetup();
			}
			
			_moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
		    }
		}
		
		if ( _evalPredicate && _moreLeft && _moreRight )
		{
		    
		    CegoQueryHelper queryHelper;
		    pJO->getPredDesc()->clearAttrCache();
		    isEvaluated = queryHelper.evalPredicate(0,
							    0,
							    0,
							    flArray,
							    offset,
							    size,
							    pJO->getPredDesc(),
							    0);		   
		}
		else
		{
		    isEvaluated = true;
		}
	    }
	    	    
	    if ( _moreLeft )
	    {
		if ( _moreRight == false )
		{
		    // make right fields null
		    CegoField *pF = flArray[offset+size-1].First();
		    while ( pF )
		    {
			CegoFieldValue nullValue;
			pF->setValue(nullValue);
			pF = flArray[offset+size-1].Next();
		    }
		}
		return true;
	    }
	    return false;
	}	
	else if ( pJO->getJoinType() == CegoJoinObject::RIGHTOUTER )
	{

	    bool isEvaluated = false;
	    
	    while ( isEvaluated == false )
	    {

		if ( _isFirst )
		{		
		    _pTCRight->distSetup(_cursorCond);
		    
		    _moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
		    
		    if ( _moreRight )
		    {
			_innerCond.setup(flArray, offset+size-1, 1);

			_pTCLeft->reset();
			
			if ( _isAttrCondValid )
			    _pTCLeft->distSetup(_innerCond);
			else
			    _pTCLeft->distSetup();

			_moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);

		    }
		    _isFirst = false;
		}
		else if ( _moreLeft )
		{
		    _moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
		    
		    if ( _moreLeft == false )
		    {
			_moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
			if ( _moreRight )
			{
			    _innerCond.setup(flArray, offset+size-1, 1);
			    
			    _pTCLeft->reset();
			    
			    if ( _isAttrCondValid )
				_pTCLeft->distSetup(_innerCond);
			    else
				_pTCLeft->distSetup();
			    
			    _moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
			}
		    }		
		}
		else if ( _moreRight )
		{
		    _moreRight = _pTCRight->nextTuple(flArray, offset+size-1, 1);
		    if ( _moreRight )
		    {
			_innerCond.setup(flArray, offset+size-1, 1);
			
			_pTCLeft->reset();
			
			if ( _isAttrCondValid )
			    _pTCLeft->distSetup(_innerCond);
			else
			    _pTCLeft->distSetup();
			
			_moreLeft = _pTCLeft->nextTuple(flArray, offset, size-1);
		    }
		}

		if ( _evalPredicate && _moreLeft && _moreRight )
		{
		    
		    CegoQueryHelper queryHelper;
		    pJO->getPredDesc()->clearAttrCache();
		    isEvaluated = queryHelper.evalPredicate(0,
							    0,
							    0,
							    flArray,
							    offset,
							    size,
							    pJO->getPredDesc(),
							    0);		   
		}
		else
		{
		    isEvaluated = true;
		}
	    }

	    if ( _moreRight )
	    {
		if ( _moreLeft == false )
		{
		    for ( int i=0; i<size-1; i++)
		    {
			// make right fields null
			CegoField *pF = flArray[offset+i].First();
			while ( pF )
			{
			    CegoFieldValue nullValue;
			    pF->setValue(nullValue);
			    pF = flArray[offset+i].Next();
			}
		    }
		}
		return true;
	    }
	    return false;
	}	    
    }

    throw Exception(EXLOC, Chain("Cannot resolve object type"));
}

void CegoDistCursor::reset()
{
    _isFirst=true;
    _moreTuple = false;

    if ( _pCO->getType() == CegoObject::VIEW )
    {
	_pSelect->reset(true);
    }
    else if ( _pCO->getType() == CegoObject::TABLE )
    {
	if ( _isLocal )
	{
	    if ( _pTC )
		_pTC->abort();
	}
	else
	{
	    if ( _pSH )
		_pDBMng->releaseSession(_pSH);
	}
    }
    else if ( _pCO->getType() == CegoObject::JOIN )
    {	
	if ( _pTCLeft )
	    _pTCLeft->reset();
	if ( _pTCRight )
	    _pTCRight->reset();
    }
    else if ( _pCO->getType() == CegoObject::SYSTEM )
    {
	if ( _pC )
	    _pC->reset();
    }
}

void CegoDistCursor::sysSetup()
{
    
    Chain tableSet = _pDBMng->getTabSetName(_tabSetId);	
    if ( _tableName == Chain(SYSTAB_TABLE_ID) )
    {
	_pGTM->getDistObjectList(tableSet, CegoObject::TABLE, _sysObjList);
    }
    else  if ( _tableName == Chain(SYSTAB_PROC_ID) )
    {
	_pGTM->getDistObjectList(tableSet, CegoObject::PROCEDURE, _sysObjList);
    }
    else  if ( _tableName == Chain(SYSTAB_VIEW_ID) )
    {
	_pGTM->getDistObjectList(tableSet, CegoObject::VIEW, _sysObjList);
    }
    else  if ( _tableName == Chain(SYSTAB_INDEX_ID) )
    {
	_pGTM->getDistObjectList(tableSet, CegoObject::INDEX, _sysObjList);
    }
    else  if ( _tableName == Chain(SYSTAB_BTREE_ID) )
    {
	_pGTM->getDistObjectList(tableSet, CegoObject::BTREE, _sysObjList);
    }
    else  if ( _tableName == Chain(SYSTAB_KEY_ID) )
    {
	_pGTM->getDistObjectList(tableSet, CegoObject::FKEY, _sysObjList);
    }
    else
    {
	_pC = _pGTM->getObjectCursor(_tabSetId, _tableName, _tableName, CegoObject::SYSTEM);
    }
}

void CegoDistCursor::joinSetup(const CegoAttrCond& attrCond)
{

    CegoJoinObject *pJO = (CegoJoinObject*)_pCO;
        
    ListT<CegoField> outerSchema;
    ListT<CegoField> innerSchema;
    
    CegoAttrCond addInnerCond;

    if ( pJO->getJoinType() == CegoJoinObject::INNER || pJO->getJoinType() == CegoJoinObject::LEFTOUTER )
    {
	outerSchema = pJO->getLeftObject()->getSchema();
	innerSchema = pJO->getRightObject()->getSchema(); 

	_cursorCond = attrCond.getFilterCond(outerSchema);
	addInnerCond = attrCond.getFilterCond(innerSchema);

    }
    else if (  pJO->getJoinType() == CegoJoinObject::RIGHTOUTER )
    {
	innerSchema = pJO->getLeftObject()->getSchema();
	outerSchema = pJO->getRightObject()->getSchema();

	_cursorCond = attrCond.getFilterCond(outerSchema);
	addInnerCond = attrCond.getFilterCond(innerSchema);
    }
    
    CegoAttrCond ac;
    CegoQueryHelper qh;
    
    CegoQueryHelper::AttrCondMatch m  = qh.checkAttrCond(ac, pJO->getPredDesc(), innerSchema, &outerSchema, 1, 0);
    
    if ( m == CegoQueryHelper::COMPLETE )
    {
	_evalPredicate = false;
	_innerCond = ac + addInnerCond;
    }
    else
    {
	_evalPredicate = true;
	_innerCond = addInnerCond;
    }

    _isAttrCondValid = true;

    if (  pJO->getJoinType() == CegoJoinObject::INNER || pJO->getJoinType() == CegoJoinObject::LEFTOUTER )
    {
	_pTCLeft->distSetup(_cursorCond);
	_pTCRight->distSetup(_innerCond);
    }
    if ( pJO->getJoinType() == CegoJoinObject::RIGHTOUTER )
    {
	_pTCRight->distSetup(_cursorCond);
	_pTCLeft->distSetup(_innerCond);
    }
}

Element* CegoDistCursor::getPlan()
{
    
    Element *pCursorPlan = new Element(XML_JOIN_ELEMENT);

    pCursorPlan->setAttribute(XML_TABLENAME_ATTR, _tableName);       
    pCursorPlan->setAttribute(XML_NAME_ATTR, _tableAlias);

    if ( _pCO->getType() == CegoObject::VIEW )
    {
	pCursorPlan->setAttribute(XML_TABLETYPE_ATTR, XML_VIEW_VALUE);
	pCursorPlan->addContent( _pSelect->getPlan() );
    }
    else if ( _pCO->getType() == CegoObject::TABLE )
    {

	pCursorPlan->setAttribute(XML_TABLETYPE_ATTR, XML_TABLE_VALUE);
	
	if ( _idxMatch == CegoAttrCond::FULL)
	{
	    pCursorPlan->setAttribute(XML_JOINSTRAT_ATTR, Chain("full index trace on ") + _cursorCond.toChain()); 
	}
	else if( _idxMatch == CegoAttrCond::PART )
	{
	    pCursorPlan->setAttribute(XML_JOINSTRAT_ATTR, Chain("index support on ") + _cursorCond.toChain()); 
	}
	else
	{
	    pCursorPlan->setAttribute(XML_JOINSTRAT_ATTR, Chain("full table scan"));
	}
	
    }
    else if ( _pCO->getType() == CegoObject::JOIN )
    {	

	CegoJoinObject *pJCO = (CegoJoinObject*)_pCO;
	
	if ( pJCO->getJoinType() == CegoJoinObject::INNER )	    
	    pCursorPlan->setAttribute(XML_TABLETYPE_ATTR, XML_INNERJOIN_VALUE);
	else if ( pJCO->getJoinType() == CegoJoinObject::LEFTOUTER)	    
	    pCursorPlan->setAttribute(XML_TABLETYPE_ATTR, XML_LEFTOUTERJOIN_VALUE);
	else if ( pJCO->getJoinType() == CegoJoinObject::RIGHTOUTER)	    
	    pCursorPlan->setAttribute(XML_TABLETYPE_ATTR, XML_RIGHTOUTERJOIN_VALUE);

	if ( _pTCLeft )
	    pCursorPlan->addContent( _pTCLeft->getPlan() );
	if ( _pTCRight )
	    pCursorPlan->addContent( _pTCRight->getPlan() );	    
    }
    else
    {
	delete pCursorPlan;
	throw Exception(EXLOC, Chain("Invalid content type"));		
    }

    return pCursorPlan;

}
