///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoBTreeManager.cc
// -------------------
// Cego index manager 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: CegoBTreeManager
// 
// Description: This class implements the btree algorithm to be used as an alternative index object.
//
// Status: 2.16
//
///////////////////////////////////////////////////////////////////////////////

// base includes
#include <lfc/Exception.h>

// cego includes
#include "CegoBTreeManager.h"
#include "CegoLockHandler.h"
#include "CegoTupleState.h"

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

CegoBTreeManager::CegoBTreeManager(CegoObjectManager *pObjMng, CegoBTreeObject *pBTO)
{
    _pDBMng = pObjMng->getDBMng();
    _modId = _pDBMng->getModId("CegoBTreeManager");
    _pObjMng = pObjMng;
    _pBTO = pBTO;

    _tabSetId = _pBTO->getTabSetId();
    _btreeName = _pBTO->getName();    
    _btreeType = _pBTO->getType();
    _btreeSchema = _pBTO->getSchema(); 

    CegoField *pF = _btreeSchema.First();
    _keyLen = 0;
    while ( pF )
    {
	_keyLen += pF->getLength();
	pF = _btreeSchema.Next();
    }
    
    int fileId = _pBTO->getDataFileId();
    int pageId = _pBTO->getDataPageId();

    if ( fileId == 0 && pageId == 0 )
    {
	throw Exception(EXLOC, Chain("BTree object is not valid"));
    }

    _rootFixed = false;
    
}

CegoBTreeManager::~CegoBTreeManager()
{
    if ( _rootFixed )
	unfixRoot();
}

void CegoBTreeManager::fixRoot()
{
    int fileId = _pBTO->getDataFileId();
    int pageId = _pBTO->getDataPageId();
   
    // cout << "Fixing root page " << fileId << " " << pageId << endl;
    _pDBMng->bufferFix(_rootBP, _tabSetId, fileId, pageId, CegoBufferPool::SYNC, _pObjMng->getLockHandler());
    _rootFixed = true;
    
}

void CegoBTreeManager::unfixRoot()
{
    // cout << "Unfixing root .." << _rootBP.getFileId() << " "  << _rootBP.getPageId() << endl;
    _pDBMng->bufferUnfix(_rootBP, true, _pObjMng->getLockHandler());    
    _rootFixed = false;
    // cout << "Unfixed .." << endl;    
}


void CegoBTreeManager::insertBTree(const CegoDataPointer& dp,
				   const CegoBTreeValue &iv,
				   int tid)
{

    
    // we have to keep the index object in buffer pool 
    CegoBufferPage bp;
    _pObjMng->getObjectWithFix(_tabSetId, _btreeName, _btreeType, *_pBTO, bp);
    
    CegoDataPointer sysEntry(bp.getFileId(), bp.getPageId(), bp.getEntryPos());

    try 
    {
	insertBTree(sysEntry, dp, iv, tid);
    }
    catch ( Exception e )
    {
	_pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());
	throw e;
    }
    _pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());
    
}

void CegoBTreeManager::insertBTree(const CegoDataPointer& sysEntry, 
				   const CegoDataPointer& dp,
				   const CegoBTreeValue &iv,
				   int tid)
{

    // cout << "Inserting index " << iv.toChain(&_btreeSchema) << endl;

    bool doAppend = true;

    StackT<CegoBufferPage> parentPageStack;
    StackT<CegoBufferPage> fixedPageStack;

    try
    {

	int fileId=0;
	int pageId=0;

	if ( _rootFixed == false )
	{
	    fileId = _pBTO->getDataFileId();
	    pageId = _pBTO->getDataPageId();	    
	}
	    
	bool isInserted = false;

	bool isFirst = true;

	while ( ! isInserted )
	{
	    
	    CegoBufferPage bp;
	    
	    if ( isFirst && _rootFixed)
	    {
		bp = _rootBP;
	    }
	    else
	    {
		_pDBMng->bufferFix(bp, _tabSetId, fileId, pageId, CegoBufferPool::SYNC, _pObjMng->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.printNode();

		traceNode.getChildPage(iv, fileId, pageId);
		
		// cout << "Tracing to child page .." << fileId << " " << pageId << endl; 
		parentPageStack.Push(bp);
	    }
	    else if ( bp.getType() == CegoBufferPage::BTREE_LEAF )
	    {
		
		CegoBTreeNode propLeft;
		propLeft.setType(CegoBTreeNode::LEAF);
		propLeft.setPtr(bp.getChunkEntry(), bp.getChunkLen());
		propLeft.setFileId(bp.getFileId());
		propLeft.setPageId(bp.getPageId());
		propLeft.setSchema(&_btreeSchema, _keyLen);
		propLeft.setNextFileId(bp.getNextFileId());
		propLeft.setNextPageId(bp.getNextPageId());
		
		if ( propLeft.addValue(iv, dp, _btreeName, _btreeType == CegoObject::PBTREE || _btreeType == CegoObject::UBTREE) == false )
		{
		    
		    // cout << "Creating new leaf page .." << endl;
		    CegoBTreeNode propRight;
		    
		    // create new 
		    CegoBufferPage newPage;
		    _pObjMng->getNewFilePage(newPage, _tabSetId, CegoObject::INDEX, true, doAppend);
		    
		    newPage.initPage(CegoBufferPage::BTREE_LEAF);
		    fixedPageStack.Push(newPage);
		    
		    propRight.setType(CegoBTreeNode::LEAF);
		    propRight.setPtr(newPage.getChunkEntry(), newPage.getChunkLen());
		    propRight.initNode();
		    propRight.setFileId(newPage.getFileId());
		    propRight.setPageId(newPage.getPageId());
		    propRight.setSchema(&_btreeSchema, _keyLen);
		    
		    // cout << "Splitting leaf page .." << endl;
		    propLeft.split(propRight);
		    
		    // cout << "Setting next page to " << newPage.getFileId() << " " << newPage.getPageId() << endl;
		    
		    if ( propLeft.getNextFileId() && propLeft.getNextPageId() )
		    {
			propRight.setNextFileId(propLeft.getNextFileId());
			propRight.setNextPageId(propLeft.getNextPageId());
			newPage.setNextFileId(propLeft.getNextFileId());
			newPage.setNextPageId(propLeft.getNextPageId());
		    }
		    
		    propLeft.setNextFileId(newPage.getFileId());
		    propLeft.setNextPageId(newPage.getPageId());
		    
		    bp.setNextFileId(newPage.getFileId());
		    bp.setNextPageId(newPage.getPageId());
		    
		    fixedPageStack.Push(bp);
		    
		    CegoBTreeValue ri = propRight.getMin();
		    
		    if ( iv.isHigher(ri, &_btreeSchema) )
		    {
			propRight.addValue(iv, dp, _btreeName, _btreeType == CegoObject::PBTREE || _btreeType == CegoObject::UBTREE);
		    }
		    else
		    {
			propLeft.addValue(iv, dp, _btreeName, _btreeType == CegoObject::PBTREE || _btreeType == CegoObject::UBTREE);
		    }
		    
		    CegoBTreeValue propValue = propLeft.getMax();
		    
		    bool isPropagated = false;
		    while ( ! isPropagated )
		    {
			// propagate newLeaf to parent node
			CegoBufferPage propPage;
			
			if ( parentPageStack.Pop(propPage) )
			{
			    
			    fixedPageStack.Push(propPage);
			    
			    CegoBTreeNode propNode;
			    propNode.setType(CegoBTreeNode::NODE);
			    propNode.setPtr(propPage.getChunkEntry(), propPage.getChunkLen());
			    propNode.setFileId(propPage.getFileId());
			    propNode.setPageId(propPage.getPageId());
			    propNode.setSchema(&_btreeSchema, _keyLen);
			    
			    if ( propNode.propagate(propValue, propLeft, propRight) == false )
			    {
				
				propLeft = propNode;
				
				// since we still have to add the right node, save it
				CegoBTreeNode addNode = propRight;
				
				// create new 
				CegoBufferPage newPage; 
				_pObjMng->getNewFilePage(newPage, _tabSetId, CegoObject::INDEX, true, doAppend);
				newPage.initPage(CegoBufferPage::BTREE_NODE);
				fixedPageStack.Push(newPage);
				
				propRight.setType(CegoBTreeNode::NODE);
				propRight.setPtr(newPage.getChunkEntry(), newPage.getChunkLen());
				propRight.initNode();
				propRight.setFileId(newPage.getFileId());
				propRight.setPageId(newPage.getPageId());
				propRight.setSchema(&_btreeSchema, _keyLen);
				
				propLeft.split(propRight);
				propValue = propLeft.getMax();
				
				propLeft.decEntries();
				
				// add new node
				CegoBTreeValue iv1 = addNode.getMin();
				CegoBTreeValue iv2 = propRight.getMin();
				
				if ( iv1.isHigher(iv2, &_btreeSchema) )
				{
				    propRight.addNode(iv1, addNode);
				}
				else
				{
				    propLeft.addNode(iv1, addNode);
				}
				
			    }
			    else
			    {
				// cout << "Node page after propagation " << endl;
				// propNode.printNode();
				
				isPropagated = true;
			    }
			    
			}
			else // root node reached, adding new root
			{
			    
			    // cout << "Adding new root page .." << endl;
			    
			    CegoBufferPage rootPage; 
			    _pObjMng->getNewFilePage(rootPage, _tabSetId, CegoObject::INDEX, true, doAppend);
			    rootPage.initPage(CegoBufferPage::BTREE_NODE);
			   
			    if ( _rootFixed )
				_rootBP = rootPage;

			    CegoBTreeNode rootNode;
			    rootNode.setType(CegoBTreeNode::NODE);
			    rootNode.setPtr(rootPage.getChunkEntry(), rootPage.getChunkLen());
			    rootNode.initNode();			
			    rootNode.setFileId(rootPage.getFileId());
			    rootNode.setPageId(rootPage.getPageId());
			    rootNode.setSchema(&_btreeSchema, _keyLen);
			    
			    rootNode.propagate(propValue, propLeft, propRight);
			    
			    // rootNode.printNode();
			    
			    _pBTO->setDataFileId(rootPage.getFileId());
			    _pBTO->setDataPageId(rootPage.getPageId());
			    
			    char *p;
			    int len;
			    unsigned long lockId = _pObjMng->claimDataPtr(_tabSetId, 
									  CegoLockHandler::WRITE, 
									  CegoBufferPool::SYNC,
									  sysEntry, p, len);
			    _pBTO->encode(p);
			    _pObjMng->releaseDataPtr(lockId, true);
			    
			    isPropagated=true;
			}
		    }		
		}
		else
		{
		    if ( isFirst == false )
			_pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());
		}
		
		isInserted = true;	    
	    }

	    isFirst = false;
	}
    }
    catch ( Exception e )
    {
	
	CegoBufferPage bp;
	while ( parentPageStack.Pop(bp) )
	{
	    _pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());
	}
	
	while ( fixedPageStack.Pop(bp) )
	{
	    _pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());
	}    
	throw e;	
    }

    // cout << "Unfixing pages .." << endl;

    // unfix node pages on stack
    CegoBufferPage bp;
    while ( parentPageStack.Pop(bp) )
    {
	if ( bp != _rootBP || _rootFixed == false)
	{
	    // cout << "Unfixing parent page .." << bp.getFileId() << " "  << bp.getPageId() << endl;
	    _pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());
	}
    }
    
    while ( fixedPageStack.Pop(bp) )
    {
	if ( bp != _rootBP || _rootFixed == false)
	{
	    // cout << "Unfixing fixed page .." << bp.getFileId() << " "  << bp.getPageId() << endl;
	    _pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());
	}
    }    
}

void CegoBTreeManager::dumpBTree()
{

    int fileId = _pBTO->getDataFileId();
    int pageId = _pBTO->getDataPageId();
    
    cout << "Root Page is " << fileId << " " << pageId << endl;
    
    if ( fileId == 0 && pageId == 0 )
    {
	throw Exception(EXLOC, Chain("BTree object is not valid"));
    }

    dumpNode(0, fileId, pageId);

}

void CegoBTreeManager::dumpNode(int level, int fileId, int pageId)
{

    CegoBufferPage bp;
    
    _pDBMng->bufferFix(bp, _tabSetId, fileId, pageId, CegoBufferPool::SYNC, _pObjMng->getLockHandler());

    CegoBTreeNode traceNode;
    traceNode.setPtr(bp.getChunkEntry(), bp.getChunkLen());	    
    traceNode.setSchema(&_btreeSchema, _keyLen);
    traceNode.setFileId(fileId);
    traceNode.setPageId(pageId);
    traceNode.setNextFileId(bp.getNextFileId());
    traceNode.setNextPageId(bp.getNextPageId());

    if ( bp.getType() == CegoBufferPage::BTREE_NODE )
    {
	traceNode.setType(CegoBTreeNode::NODE);	
	traceNode.printNode(level);
	
	int childFid, childPid;
	
	traceNode.reset();
	while ( traceNode.nextChildPointer(childFid, childPid) )
	{
	    dumpNode(level + 3, childFid, childPid);  
	}
    }
    else
    {
	traceNode.setType(CegoBTreeNode::LEAF);
	traceNode.printNode(level);
    }
   
    _pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());

}


void CegoBTreeManager::deleteBTree(const CegoBTreeValue &iv,
				   const CegoDataPointer& dp)
{    


    int fileId = _pBTO->getDataFileId();
    int pageId = _pBTO->getDataPageId();

    if ( fileId == 0 && pageId == 0 )
    {
	throw Exception(EXLOC, Chain("BTree object is not valid"));
    }
    
    bool isDeleted = false;

    StackT<CegoBufferPage> parentPageStack;
    StackT<CegoBufferPage> fixedPageStack;
    while ( ! isDeleted )
    {
   
	CegoBufferPage bp;
	
	_pDBMng->bufferFix(bp, _tabSetId, fileId, pageId, CegoBufferPool::SYNC, _pObjMng->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.printNode();

	    traceNode.getChildPage(iv, fileId, pageId);

	    // cout << "Tracing to child page .." << fileId << " " << pageId << endl; 

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

	    CegoBTreeNode leaf;
	    leaf.setType(CegoBTreeNode::LEAF);
	    leaf.setPtr(bp.getChunkEntry(), bp.getChunkLen());
	    leaf.setFileId(bp.getFileId());
	    leaf.setPageId(bp.getPageId());
	    leaf.setSchema(&_btreeSchema, _keyLen);
	    leaf.setNextFileId(bp.getNextFileId());
	    leaf.setNextPageId(bp.getNextPageId());
	    	
	    leaf.deleteValue(iv, dp);
	    
	    _pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());

	    isDeleted = true;	    
	}
    }

    // unfix node pages on stack
    CegoBufferPage bp;
    while ( parentPageStack.Pop(bp) )
    {
	_pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());
    }
}

void CegoBTreeManager::freeBTree()
{
    freeNodePages(_pBTO->getDataFileId(), _pBTO->getDataPageId());
}


void CegoBTreeManager::freeNodePages(int fileId, int pageId)
{
    int pageCount = 0;

    CegoBufferPage bp;    
    _pDBMng->bufferFix(bp, _tabSetId, fileId, pageId, CegoBufferPool::SYNC, _pObjMng->getLockHandler());

    if ( bp.getType() == CegoBufferPage::BTREE_NODE )
    {

	CegoBTreeNode traceNode;
	traceNode.setPtr(bp.getChunkEntry(), bp.getChunkLen());	    
	traceNode.setSchema(&_btreeSchema, _keyLen);
	traceNode.setFileId(fileId);
	traceNode.setPageId(pageId);	
	traceNode.setType(CegoBTreeNode::NODE);	

	int childFid, childPid;
	
	traceNode.reset();
	while ( traceNode.nextChildPointer(childFid, childPid) )
	{	    
	    freeNodePages(childFid, childPid);
	}
    }

    _pDBMng->bufferRelease(bp, _pObjMng->getLockHandler());

}

int CegoBTreeManager::getNumPages()
{
    return countNodePages(_pBTO->getDataFileId(), _pBTO->getDataPageId());
}

int CegoBTreeManager::countNodePages(int fileId, int pageId)
{

    int pageCount = 0;

    CegoBufferPage bp;    
    _pDBMng->bufferFix(bp, _tabSetId, fileId, pageId, CegoBufferPool::SYNC, _pObjMng->getLockHandler());

    if ( bp.getType() == CegoBufferPage::BTREE_NODE )
    {

	CegoBTreeNode traceNode;
	traceNode.setPtr(bp.getChunkEntry(), bp.getChunkLen());	    
	traceNode.setSchema(&_btreeSchema, _keyLen);
	traceNode.setFileId(fileId);
	traceNode.setPageId(pageId);	
	traceNode.setType(CegoBTreeNode::NODE);	

	int childFid, childPid;
	
	traceNode.reset();
	while ( traceNode.nextChildPointer(childFid, childPid) )
	{
	    pageCount += countNodePages(childFid, childPid);
	}
    }
    else
    {
	pageCount++;
    }
   
    _pDBMng->bufferUnfix(bp, true, _pObjMng->getLockHandler());

    return pageCount;
}


