///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoGroupSpace.cc
// ------------------
// Cego group space 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: CegoGroupSpace
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

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

// cego includes
#include "CegoGroupSpace.h"
#include "CegoDataType.h"
#include "CegoTableManager.h"
#include "CegoGroupEntry.h"
#include "CegoTypeConverter.h"

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

CegoGroupSpace::CegoGroupSpace()
{
    _pAVL = new AVLTreeT<CegoGroupNode>;
    _pCountAgg = new CegoAggregation();
}

CegoGroupSpace::~CegoGroupSpace()
{
    if ( _pCountAgg )
	delete _pCountAgg;
}

void CegoGroupSpace::initGroupSpace(const ListT<CegoField>& schema, ListT<CegoAggregation*>& aggList, long maxOrderSize)
{
    _aggList = aggList;
    _maxOrderSize = maxOrderSize;
    _orderSize = 0;
    _groupSchema.Empty();
    _aggSchema.Empty();

    int id=1;
    
    CegoField* pF = schema.First();
    while ( pF )
    {
	pF->setId(id);

	_groupSchema.Insert(*pF);

	id++;
	pF = schema.Next();
    }

    /* we add count aggregation in any case */
    _pCountAgg->setAggregationId(_aggList.Size() + 1);
    _aggList.Insert( _pCountAgg );

    CegoAggregation** pAgg = _aggList.First();
    while ( pAgg )
    {
	CegoField f;
	f.setId(id);
	(*pAgg)->setAggregationId(id);

	f.setType(INT_TYPE);
	f.setAttrName((*pAgg)->toChain());

	_groupSchema.Insert(f);
	_aggSchema.Insert(f);
	
	id++;
	pAgg = _aggList.Next();
    }

    _groupingOffset = schema.Size() + 1;

    int i=1;
    pF = _groupSchema.First();
    while ( pF )
    {
	if ( i < _groupingOffset )
	    pF->setTableAlias("GROUPING");
	else
	{
	    if ( _aggList[i-_groupingOffset]->getType() == CegoAggregation::AVG )
	    {
		pF->setTableAlias(AVG_GROUP);
	    }
	    else if ( _aggList[i-_groupingOffset]->getType() == CegoAggregation::MIN )
	    {
		pF->setTableAlias(MIN_GROUP);
	    }
	    else if ( _aggList[i-_groupingOffset]->getType() == CegoAggregation::MAX )
	    {
		pF->setTableAlias(MAX_GROUP);
	    }
	    else if ( _aggList[i-_groupingOffset]->getType() == CegoAggregation::SUM )
	    {
		pF->setTableAlias(SUM_GROUP);
	    }
	    else
	    {
		pF->setTableAlias(COUNT_GROUP);
	    }
	}
	
	i++;
	pF = _groupSchema.Next();
    }
}

void CegoGroupSpace::resetGroupSpace()
{    
    _pAVL->Empty();
    _orderSize = 0;
}

ListT<CegoField> CegoGroupSpace::getSchema()
{
    return _groupSchema;
}

void CegoGroupSpace::insertTuple(ListT<CegoField>& addKey, ListT<CegoField>& dataTuple)
{
    
    int id=1;
    CegoField* pF = addKey.First();
    while (pF)
    {
	pF->setId(id);
	id++;
	pF = addKey.Next();
    }
        
    CegoGroupNode n(addKey);
    CegoGroupNode* pN = _pAVL->Find(n);
    if ( pN )
    {
	ListT<CegoField> newGroupValue = aggGrouping(pN->getGrouping(), dataTuple);
	pN->setGrouping(newGroupValue);
    }
    else
    {	
	ListT<CegoField> groupedValues = initGrouping(dataTuple);
	CegoGroupNode nn(addKey, groupedValues);
	_orderSize += sizeof(nn);

	if ( _orderSize > _maxOrderSize )
	{
	    throw Exception(EXLOC, "Order size exceeded");
	}
	
	_pAVL->Insert(nn);	

    }
}

CegoGroupCursor* CegoGroupSpace::getCursor()
{
    CegoGroupCursor *pGC = new CegoGroupCursor(_groupSchema, _pAVL);
    return pGC;
}

long CegoGroupSpace::numAllocated() const
{
    return _orderSize;
}

ListT<CegoField> CegoGroupSpace::initGrouping(ListT<CegoField>& dataTuple)
{

    ListT<CegoField> initValue;

    CegoAggregation** pAgg = _aggList.First();
    int i=_groupingOffset;
    while ( pAgg )
    {
	CegoField f;
	f.setId(i);
        
	switch ( (*pAgg)->getType() )
	{	    
	case CegoAggregation::COUNT:
	{
	    f.setValue(CegoFieldValue(INT_TYPE, Chain("1")));
	    break;
	}	
	case CegoAggregation::SUM:
	case CegoAggregation::AVG:
	case CegoAggregation::MIN:
	case CegoAggregation::MAX:
	{	    
	    f.setValue( getValueForAgg(*pAgg, dataTuple));
	    break;
	}	
	}

	initValue.Insert(f);
	
	i++;
	pAgg = _aggList.Next();
    }
    
    return initValue;
}

ListT<CegoField> CegoGroupSpace::aggGrouping(ListT<CegoField>& aggValues, ListT<CegoField>& dataTuple)
{
    
    CegoAggregation** pAgg = _aggList.First();
    CegoField *pF = aggValues.First();

    while ( pAgg && pF )
    {
	(*pAgg)->setFieldValue(pF->getValue());
	pAgg = _aggList.Next();
	pF = aggValues.Next();
    }

    ListT<CegoField> aggValue;

    int i=0;
    pAgg = _aggList.First();
    while ( pAgg  )
    {

	CegoQueryHelper qh;
	qh.aggregateTuple(dataTuple, *pAgg );
	
	CegoField f;
	f.setAttrName( _aggSchema[i].getAttrName());
	f.setValue( (*pAgg)->getFieldValue() );
	f.setId(i + _groupingOffset);
	i++;
	aggValue.Insert(f);
	pAgg = _aggList.Next();
    }

    return aggValue;
    
}

CegoFieldValue CegoGroupSpace::getValueForAgg(CegoAggregation* pAgg, ListT<CegoField>& dataTuple)
{
	
    CegoField f;
    CegoFieldValue fv;

    CegoExpr *pAE = pAgg->getExpr();
    
    if ( pAE == 0 )
    {
	f.setAttrName(Chain("*"));
	fv.setType(INT_TYPE);
    }
    else
    {

	pAE->setFieldListArray( &dataTuple, 1 );
	fv = pAE->evalFieldValue();

	if ( fv.getValue() == 0 && pAgg->getType() != CegoAggregation::MIN )
	{
	    switch ( fv.getType() )
	    {
	    case INT_TYPE:
	    {			
		fv = CegoFieldValue(INT_TYPE, Chain("0"));
		break;
	    }
	    case VARCHAR_TYPE:
	    {
		fv = CegoFieldValue(VARCHAR_TYPE, Chain(""));
		break;
	    }
	    case DECIMAL_TYPE:
	    {
		fv = CegoFieldValue(DECIMAL_TYPE, Chain("0.0"));
		break;		   
	    }
	    case FIXED_TYPE:
	    {
		fv = CegoFieldValue(FIXED_TYPE, Chain("0.0"));
		break;		   
	    }
	    default:
	    {
		Chain msg = Chain("Aggregation not supported on datatype <") + CEGO_TYPE_MAP[ (int) fv.getType()] + Chain(">");
		throw Exception(EXLOC, msg);
		break;
	    }
	    }
	}
    }
    
    return fv;
}



