///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoBTreeCursor.cc
// ------------------
// Cego index cursor class 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: CegoBTreeCursor
// 
// Description: Index cursors are used for reading tables based on an available index 
//              for the given attribute condition
//
// Status: QG-2.13
//
///////////////////////////////////////////////////////////////////////////////

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

// cego includes
#include "CegoBTreeCursor.h"
#include "CegoTupleState.h"
#include "CegoBTreeNode.h"
#include "CegoBTreeManager.h"

CegoBTreeCursor::CegoBTreeCursor()
{
}

CegoBTreeCursor::CegoBTreeCursor(CegoTableManager *pTabMng, int tabSetId, const Chain& btreeName, CegoObject::ObjectType type, CegoAttrCond* pAttrCond, bool ignoreTouched, bool readUncommitted)
{
    _pTabMng = pTabMng;
    _pDBMng = pTabMng->getDBMng();
    _btreeName = btreeName;
    _type = type;
    _pAttrCond = pAttrCond;

    _dataLock = 0;

    _tabSetId = tabSetId;
    _ignoreTouched = ignoreTouched;
    _readUncommitted = readUncommitted;
    _cursorCached = false;
    _eoc=true;
}

CegoBTreeCursor::~CegoBTreeCursor()
{
    abort();
}

bool CegoBTreeCursor::getFirst(ListT<CegoField>& fl, CegoDataPointer& dp)
{

    _eoc=false;

    if ( fl.isEmpty() )
    {
	throw Exception(EXLOC, "Empty field list");
    }

    if ( _cursorCached == false )
    {
	CegoBTreeObject btoe;
	
	_pTabMng->getObject(_tabSetId, _btreeName, _type, btoe);

	_btreeSchema = btoe.getSchema();

	CegoField *pF = _btreeSchema.First();
	_keyLen = 0;
	while ( pF )
	{
	    _keyLen += pF->getLength();
	    pF = _btreeSchema.Next();
	}

	_rootFileId = btoe.getDataFileId();
	_rootPageId = btoe.getDataPageId();
	
	Chain tabName = btoe.getTabName();
	
	_cursorCached = true;

	// CegoBTreeManager btm(_pTabMng);
	// btm.dumpBTree(btoe);
    }

    bool found = false;

    if (_pAttrCond)
    { 

	if ( _pAttrCond->getPrimaryCompMode() == CegoAttrComp::BTWN )
	    _pAttrCond->setPrimaryComparison( MORE_EQUAL_THAN );

	switch ( _pAttrCond->getPrimaryComparison() )
	{
	case NOT_EQUAL:
	case LESS_THAN:
	case LESS_EQUAL_THAN:
	{
	    // go to the beginning

	    bool leafFound = false;
	    _curFileId = _rootFileId;
	    _curPageId = _rootPageId;

	    CegoBTreeValue iv;

	    while ( leafFound == false )
	    {

		CegoBufferPage bp;
		
		_pDBMng->bufferFix(bp, _tabSetId, _curFileId, _curPageId, CegoBufferPool::SYNC, _pTabMng->getLockHandler());
				
		if ( bp.getType() == CegoBufferPage::BTREE_NODE )
		{
		    
		    CegoBTreeNode traceNode;
		    traceNode.setType(CegoBTreeNode::NODE);
		    traceNode.setPtr(bp.getChunkEntry(), bp.getChunkLen());	    
		    traceNode.setSchema(&_btreeSchema, _keyLen);
		    traceNode.setFileId(bp.getFileId());
		    traceNode.setPageId(bp.getPageId());
		    
		    traceNode.reset();
		    traceNode.nextChildPointer(_curFileId, _curPageId);

		    _parentPageStack.Push(bp);
		}
		else if ( bp.getType() == CegoBufferPage::BTREE_LEAF )
		{

		    _curPage = bp;
		    _curLeaf.setType(CegoBTreeNode::LEAF);
		    _curLeaf.setPtr(_curPage.getChunkEntry(), _curPage.getChunkLen());
		    _curLeaf.setFileId(_curPage.getFileId());
		    _curLeaf.setPageId(_curPage.getPageId());
		    _curLeaf.setSchema(&_btreeSchema, _keyLen);

		    leafFound=true;

		    _curLeaf.reset();
		    bool hasValue = _curLeaf.nextValue(iv, dp);
		}
	    }
	    	    
	    CegoComparison comp = compValue(iv);
	    
	    if ( comp == LESS_THAN
		 || ( comp == EQUAL && _pAttrCond->getPrimaryComparison() == LESS_EQUAL_THAN )
		 || ( comp != EQUAL && _pAttrCond->getPrimaryComparison() == NOT_EQUAL ) )
	    {	       
		found = true;
	    }
	    else if ( _pAttrCond->getPrimaryComparison() == NOT_EQUAL )
	    {
		return getNext(fl, dp);
	    }
	    	    
	    break;
	}
	case EQUAL:
	case MORE_EQUAL_THAN:
	{

	    bool leafFound = false;

	    _curFileId = _rootFileId;
	    _curPageId = _rootPageId;

	    while ( leafFound == false )
	    {
		CegoBufferPage bp;
		_pDBMng->bufferFix(bp, _tabSetId, _curFileId, _curPageId, CegoBufferPool::SYNC, _pTabMng->getLockHandler());
				
		if ( bp.getType() == CegoBufferPage::BTREE_NODE )
		{
		    
		    CegoBTreeNode traceNode;
		    traceNode.setType(CegoBTreeNode::NODE);
		    traceNode.setPtr(bp.getChunkEntry(), bp.getChunkLen());	    
		    traceNode.setSchema(&_btreeSchema, _keyLen);
		    traceNode.setFileId(bp.getFileId());
		    traceNode.setPageId(bp.getPageId());

		    traceNode.reset();
		   
		    bool goOn = traceNode.nextChildPointer(_curFileId, _curPageId);
		    while ( goOn )
		    {
			CegoBTreeValue iv;
			if ( traceNode.rightChild(iv) )
			{
			    CegoComparison comp = compValue(iv);
			    
			    if ( comp == EQUAL
				 || comp == MORE_THAN )
			    {	       
				goOn = false;
			    }
			    else
			    {
				goOn = traceNode.nextChildPointer(_curFileId, _curPageId);
			    }
			}
			else
			{
			    goOn = false;
			}			
		    }				
		    _parentPageStack.Push(bp); 
		}
		else if ( bp.getType() == CegoBufferPage::BTREE_LEAF )
		{
		
		    _curPage = bp;
		    _curLeaf.setType(CegoBTreeNode::LEAF);
		    _curLeaf.setPtr(_curPage.getChunkEntry(), _curPage.getChunkLen());
		    _curLeaf.setFileId(_curPage.getFileId());
		    _curLeaf.setPageId(_curPage.getPageId());
		    _curLeaf.setSchema(&_btreeSchema, _keyLen);

		    leafFound=true;

		    CegoBTreeValue iv;
		    
		    _curLeaf.reset();
		    bool goOn = _curLeaf.nextValue(iv, dp);
		    while ( goOn )
		    {
			CegoComparison comp = compValue(iv);			
			if ( comp == EQUAL || comp == MORE_THAN )
			{	       
			    goOn = false;			    
			    if ( comp == MORE_THAN &&  _pAttrCond->getPrimaryComparison() == EQUAL )
				found = false;
			    else
				found = true;			    
			}
			else
			{
			    goOn = _curLeaf.nextValue(iv, dp);
			}
		    }
		}
	    }	    	    
	    break;	    
	}
	case MORE_THAN:
	{

	    // go to the most right

	    bool leafFound = false;
	    _curFileId = _rootFileId;
	    _curPageId = _rootPageId;


	    while ( ! found )
	    {

		CegoBTreeValue startVal;

		while ( leafFound == false )
		{
		    
		    CegoBufferPage bp;
		    
		    _pDBMng->bufferFix(bp, _tabSetId, _curFileId, _curPageId, CegoBufferPool::SYNC, _pTabMng->getLockHandler());
		    
		    
		    if ( bp.getType() == CegoBufferPage::BTREE_NODE )
		    {
		    
			CegoBTreeNode traceNode;
			traceNode.setType(CegoBTreeNode::NODE);
			traceNode.setPtr(bp.getChunkEntry(), bp.getChunkLen());	    
			traceNode.setSchema(&_btreeSchema, _keyLen);
			traceNode.setFileId(bp.getFileId());
			traceNode.setPageId(bp.getPageId());
			
			
			traceNode.getChildPage(startVal, _curFileId, _curPageId);
		    

		    }
		    else if ( bp.getType() == CegoBufferPage::BTREE_LEAF )
		    {
			
			leafFound=true;
			
			CegoBTreeNode minLeaf;
			minLeaf.setType(CegoBTreeNode::LEAF);
			minLeaf.setPtr(bp.getChunkEntry(), bp.getChunkLen());
			minLeaf.setFileId(bp.getFileId());
			minLeaf.setPageId(bp.getPageId());
			minLeaf.setSchema(&_btreeSchema, _keyLen);
			
			// dp = minLeaf.getNextValue(startVal);
			
		    }
		}
		
	    }
	    break;	    
	}       	
	}
    }
    else
    {
	
	// go to the beginning
	
	bool leafFound = false;
	_curFileId = _rootFileId;
	_curPageId = _rootPageId;
	
	CegoBTreeValue minVal;
	
	while ( leafFound == false )
	{
	    
	    CegoBufferPage bp;
	    
	    _pDBMng->bufferFix(bp, _tabSetId, _curFileId, _curPageId, CegoBufferPool::SYNC, _pTabMng->getLockHandler());

	    
	    if ( bp.getType() == CegoBufferPage::BTREE_NODE )
	    {
		
		CegoBTreeNode traceNode;
		traceNode.setType(CegoBTreeNode::NODE);
		traceNode.setPtr(bp.getChunkEntry(), bp.getChunkLen());	    
		traceNode.setSchema(&_btreeSchema, _keyLen);
		traceNode.setFileId(bp.getFileId());
		traceNode.setPageId(bp.getPageId());
				    	
		traceNode.reset();
		traceNode.nextChildPointer(_curFileId, _curPageId);

		_parentPageStack.Push(bp);

	    }
	    else if ( bp.getType() == CegoBufferPage::BTREE_LEAF )
	    {
		
		_curPage = bp;
		_curLeaf.setType(CegoBTreeNode::LEAF);
		_curLeaf.setPtr(bp.getChunkEntry(), bp.getChunkLen());
		_curLeaf.setFileId(bp.getFileId());
		_curLeaf.setPageId(bp.getPageId());
		_curLeaf.setSchema(&_btreeSchema, _keyLen);


		CegoBTreeValue iv;
		
		_curLeaf.reset();
		bool goOn = _curLeaf.nextValue(iv, dp);
		
		
		leafFound=true;
	    }
	   	    
	    found = true;
	}
    }

    // cout << "After search ..." << dp  << endl;

    if (found )
    {

	char* p;
	int len;
	
	if (_dataLock)
	{
	    _pTabMng->releaseDataPtr(_dataLock, false);
	    _dataLock = 0;
	}
       
	_dataLock = _pTabMng->claimDataPtr(_tabSetId, CegoLockHandler::READ, CegoBufferPool::SYNC, dp, p, len);
	
	// skipping tid
	
	int tid;
	int tastep;
	CegoTupleState ts;

	memcpy(&tid, p, sizeof(int));

	char* ptastep = p + sizeof(int);
	memcpy(&tastep, ptastep, sizeof(int));

	char* pts = p + sizeof(int) + sizeof(int);
	memcpy(&ts, pts, sizeof(CegoTupleState));
	
	if (tid != 0)
	{
	    if ( _ignoreTouched )
	    {			
		if ( ts == INSERTED 
		     && tid == _pTabMng->getTID(_tabSetId) 
		     && tastep < _pTabMng->getTAStep(_tabSetId) )
		{
		    CegoTupleState ts;
		    _qh.decodeFVL(fl, p, len, tid, tastep, ts);
		    return true;   
		}
		else
		{
		    return getNext(fl,dp);
		}
	    }
	    else
	    {
		
		if ( ( _readUncommitted == true 
		       && ts == INSERTED )
		     || ( _readUncommitted == false 
			  && (( ts == INSERTED && tid == _pTabMng->getTID(_tabSetId)) 
			      || ( ts == DELETED && tid != _pTabMng->getTID(_tabSetId)))))
		{
		    _qh.decodeFVL(fl, p, len, tid, tastep, ts);
		    return true;   
		}
		else
		{
		    return getNext(fl,dp);		    
		}
	    }
	}
	else
	{
	    _qh.decodeFVL(fl, p, len, tid, tastep, ts);
	    return true;
	}
    }
    else
    {
	_eoc=true;
    }
    
    return false;
}

bool CegoBTreeCursor::getNext(ListT<CegoField>& fl, CegoDataPointer& dp)
{

    if ( _eoc )
	return false;

    if ( fl.isEmpty() )
    {
	throw Exception(EXLOC, "Empty field list");
    }

    CegoBTreeValue iv;
    
    bool moreValue = _curLeaf.nextValue(iv, dp);

    while ( moreValue == false && _eoc == false )
    {

	int fileId = _curPage.getNextFileId();
	int pageId = _curPage.getNextPageId();

	_pDBMng->bufferUnfix(_curPage, true, _pTabMng->getLockHandler());	

	if ( fileId && pageId )
	{
	    _pDBMng->bufferFix(_curPage, _tabSetId, fileId, pageId, CegoBufferPool::SYNC, _pTabMng->getLockHandler());
	    
	    _curLeaf.setPtr(_curPage.getChunkEntry(), _curPage.getChunkLen());
	    _curLeaf.setFileId(_curPage.getFileId());
	    _curLeaf.setPageId(_curPage.getPageId());
	    _curLeaf.setSchema(&_btreeSchema, _keyLen);
	    
	    
	    _curLeaf.reset();
	    moreValue = _curLeaf.nextValue(iv, dp);
	}
	else
	{
	    _eoc = true;
	}
    }


    do {

	
	bool attrMatch = false;

	// get next available entry from leaf page
	
	CegoComparison comp = compValue(iv);
        
	if (_pAttrCond)
	{

	    if ( _pAttrCond->getPrimaryCompMode() == CegoAttrComp::BTWN )
	    {
		_pAttrCond->setPrimaryComparison(LESS_EQUAL_THAN);
	    }
	  	  
	    switch (_pAttrCond->getPrimaryComparison())
	    {
		
	    case EQUAL:
	    {	  
		switch (comp)
		{
		case EQUAL:
		    attrMatch=true;
		    break;
		default:
		    break;
		}
		break;
	    }
	    case NOT_EQUAL:
	    {
		switch (comp)
		{
		case LESS_THAN:
		    attrMatch=true;
		    break;
		case MORE_THAN:
		    attrMatch=true;
		    break;
		case EQUAL:
		    return getNext(fl, dp);
		default:
		    break;
		}
		break;
	    }
	    case LESS_THAN:
	    {
		
		switch (comp)
		{
		case LESS_THAN:
		    attrMatch=true;
		    break;
		default:
		    break;
		}
		break;
	    }
	    case MORE_THAN:	     
	    {
		switch (comp)
		{
		case MORE_THAN:
		    attrMatch=true;
		    break;
		default:
		    break;		
		}	
		break;       
	    }
	    case LESS_EQUAL_THAN:
	    {
		switch (comp)
		{
		case EQUAL:
		    attrMatch=true;
		    break;
		case LESS_THAN:
		    attrMatch=true;
		    break;
		default:
		    break;
		}
		break;
	    }
	    
	    case MORE_EQUAL_THAN:
	    {
		switch (comp)
		{
		case EQUAL:
		    attrMatch=true;
		    break;
		case MORE_THAN:
		    attrMatch=true;
		    break;
		default:
		    break;
		}
		break;
	    }
	    }	    
	}
	else
	{
	    attrMatch = true;
	}
	
	if ( attrMatch )
	{
	    
	    char* p;
	    int len;
	    	    
	    // dp = _ie.getData();
	    
	    if (_dataLock)
	    {
		_pTabMng->releaseDataPtr(_dataLock, false);
		_dataLock = 0;
	    }
	    _dataLock = _pTabMng->claimDataPtr(_tabSetId, CegoLockHandler::READ, CegoBufferPool::SYNC, dp, p, len);
	    
	    // skipping tid
	    
	    int tid;
	    int tastep;
	    CegoTupleState ts;
	    memcpy(&tid, p, sizeof(int));

	    char* ptastep = p + sizeof(int);
	    memcpy(&tastep, ptastep, sizeof(int));

	    char* pts = p + sizeof(int) + sizeof(int);
	    memcpy(&ts, pts, sizeof(CegoTupleState));
	    
	    if (tid != 0)
	    {
		if ( _ignoreTouched == true )
		{
		    if ( ts == INSERTED 
			 && tid == _pTabMng->getTID(_tabSetId) 
			 && tastep < _pTabMng->getTAStep(_tabSetId) )
		    {
			CegoTupleState ts;
			_qh.decodeFVL(fl, p, len, tid, tastep, ts);
			return true;   
		    }
		    else
		    {		
			// ignore entry
		    }
		}
		else
		{

		    if ( ( _readUncommitted == true 
			   && ts == INSERTED )
			 || ( _readUncommitted == false 
			      && (( ts == INSERTED && tid == _pTabMng->getTID(_tabSetId)) 
				  || ( ts == DELETED && tid != _pTabMng->getTID(_tabSetId)))))
		    {
			CegoTupleState ts;
			_qh.decodeFVL(fl, p, len, tid, tastep, ts);
			return true;
		    }
		    else
		    {
			// ignore entry
		    }
		}
	    }
	    else
	    {
		
		CegoTupleState ts;
		_qh.decodeFVL(fl, p, len, tid, tastep, ts);
		return true;
	    }
	}
	else
	{
	    abort();
	    _eoc = true;
	    return false;
	}
    }  while (1);
}

void CegoBTreeCursor::abort()
{
    // unfix node pages on stack
    CegoBufferPage bp;
    while ( _parentPageStack.Pop(bp) )
    {
	// cout << "Unfixing parent page " << bp.getFileId() << " " << bp.getPageId() << endl;
	_pDBMng->bufferUnfix(bp, true, _pTabMng->getLockHandler());
    }

    while ( _fixedPageStack.Pop(bp) )
    {
	// cout << "Unfixing new page " << bp.getFileId() << " " << bp.getPageId() << endl;
	_pDBMng->bufferUnfix(bp, true, _pTabMng->getLockHandler());
    }       
}

void CegoBTreeCursor::reset()
{
    abort();
}

CegoComparison CegoBTreeCursor::compValue(const CegoBTreeValue& val)
{

    // cout << "Comparing value " << val.toChain(_btreeSchema) << endl;

    ListT<CegoFieldValue> fvl = val.valueToFVL(_btreeSchema);

    CegoField* pF = _btreeSchema.First();
    CegoFieldValue *pFV = fvl.First();

    while ( pF ) 
    {

	CegoAttrComp* pAC = _pAttrCond->getAttrCompSet().First();

	while ( pAC )
	{
	    // cout << "Comp is " << pAC->toChain() << endl;

	    if  ( (Chain)pAC->getAttrName() == (Chain)pF->getAttrName()  )
	    {	       
		if ( *pFV < pAC->getFieldValue() )
		{
		    return LESS_THAN;
		}
		
		if ( pAC->getCompMode() == CegoAttrComp::VAL
		     || pAC->getCompMode() == CegoAttrComp::ATTR )
		{
		    if ( *pFV > pAC->getFieldValue() )
		    {
			return MORE_THAN;
		    }
		}
		else if ( pAC->getCompMode() == CegoAttrComp::BTWN )
		{
		    if ( *pFV > pAC->getFieldValue2() )
		    {
			return MORE_THAN;
		    }
		}		
	    }
	    pAC = _pAttrCond->getAttrCompSet().Next();
	}		


	pF = _btreeSchema.Next();
	pFV = fvl.Next();

    }

    return EQUAL;

}

