///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoImpInStream.cc 
// ------------------
// Cego import instream implementation
//     
// Design and Implementation by Bjoern Lemke               
//     
// (C)opyright 2000-2010 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: CegoImpInStream
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

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

// cego includes
#include "CegoImpInStream.h"
#include "CegoXMLdef.h"
#include "CegoField.h"
#include "CegoObject.h"
#include "CegoDistManager.h"
#include "CegoTypeConverter.h"

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

CegoImpInStream::CegoImpInStream(const Chain& tableSet, CegoDistManager* pGTM, bool doLogging)
{
    _pGTM = pGTM;
    _tableSet = tableSet;
    _pDBMng = _pGTM->getDBMng();
    _mode = IMP_ALL;
    _doLogging = doLogging;
    _modId = _pDBMng->getModId("CegoImpInStream");
}

CegoImpInStream::CegoImpInStream(const Chain& tableSet, const Chain& tableName, CegoDistManager* pGTM, bool doLogging)
{
    _pGTM = pGTM;
    _tableSet = tableSet;
    _impTable = tableName;
    _pDBMng = _pGTM->getDBMng();
    _mode = IMP_TABLE;
    _doLogging = doLogging;
    _modId = _pDBMng->getModId("CegoImpInStream");
}

CegoImpInStream::~CegoImpInStream()
{
    if  ( _bp.isFixed() )
	_pDBMng->bufferUnfix(_bp, true, _pGTM->getLockHandler());  
}

void CegoImpInStream::putFirst(Element* pParent, const Chain& name, const ListT<Attribute>& attrList, ListT<char*> dataList)
{

    _isFirst = true;
    if  ( _bp.isFixed() )
    {
	_pDBMng->bufferUnfix(_bp, true, _pGTM->getLockHandler());  
    }

    Chain tabName = pParent->getAttributeValue(XML_NAME_ATTR);
    
    if ( _mode == IMP_ALL ||  ( _mode == IMP_TABLE && tabName == _impTable) )
    {
	
	_pDBMng->log(_modId, Logger::NOTICE, Chain("Importing table ") + tabName + Chain(" ..."));
	
	ListT<Element*> schemaList = pParent->getChildren(XML_SCHEMA_ELEMENT);
	
	Element **pSchemaElement = schemaList.First();
	
	_schema.Empty();
	
	if ( pSchemaElement )
	{
	    ListT<Element*> colList = (*pSchemaElement)->getChildren(XML_COL_ELEMENT);
	    
	    Element **pCol = colList.First();
	    while ( pCol ) 
	    {
		
		Chain colName = (*pCol)->getAttributeValue(XML_COLNAME_ATTR);
		Chain colType = (*pCol)->getAttributeValue(XML_COLTYPE_ATTR);
		Chain colSize = (*pCol)->getAttributeValue(XML_COLSIZE_ATTR);
		Chain colNullable = (*pCol)->getAttributeValue(XML_COLNULLABLE_ATTR);
		bool isNullable;
		if ( colNullable == Chain(XML_TRUE_VALUE) ) 
		    isNullable = true;
		else
		    isNullable = false;

		
		CegoTypeConverter tc;
		CegoDataType type = tc.getTypeId( colType );
		
		CegoFieldValue defValue;
		Chain colDefValue = (*pCol)->getAttributeValue(XML_COLDEFVALUE_ATTR);
		
		if ( colDefValue != Chain(""))
		    defValue = CegoFieldValue(type, colDefValue);
				
		CegoField f(CegoField(tabName, tabName, colName, type, colSize.asInteger(), defValue, isNullable));		
		_schema.Insert(f);
		
		pCol = colList.Next();
	    }
	    
	    ListT<CegoField> idxList;
	    _pGTM->createDistDataTable(_tableSet, tabName, CegoObject::TABLE, _schema, idxList);		
	    _pGTM->setAppend(true);

	}
	insertData(tabName, attrList, dataList);
    }
}

void CegoImpInStream::putNext(Element* pParent, const Chain& name, const ListT<Attribute>& attrList, ListT<char*> dataList)
{
    Chain tabName = pParent->getAttributeValue(XML_NAME_ATTR);
    if ( _mode == IMP_ALL || ( _mode == IMP_TABLE && tabName == _impTable) )
    {
	insertData(tabName, attrList, dataList);
    }
}

void CegoImpInStream::insertData(const Chain& tabName, const ListT<Attribute>& attrList, ListT<char*> dataList)
{
    // _pDBMng->log(_modId, Logger::DEBUG, Chain("Inserting data into ") + tabName + Chain(" ..."));

    if ( attrList.Size() == 0 )
	return;
    
    ListT<CegoField> fvl;

    CegoField *pF = _schema.First();
    while ( pF ) 
    {

	Chain colName = pF->getAttrName();
	
	Attribute *pAttr = attrList.Find( Attribute( colName ) );

	if ( pAttr )
	{
	    Chain colVal = pAttr->getValue();
	    
	    CegoTypeConverter tc;
	    if ( pF->getType() != BLOB_TYPE )
	    {
		int len = tc.getTypeLen(pF->getType(), colVal);
		CegoField f(CegoField(pF->getTableName(), pF->getTableAlias(), colName, pF->getType(), len));
		CegoFieldValue fv(pF->getType(), colVal);
		f.setValue(fv);
		fvl.Insert(f);
	    }
	    else
	    {
		
		int blobPos = colVal.cutTrailing(XML_BLOBPREFIX).asInteger();
		
		int tabSetId = _pDBMng->getTabSetId(_tableSet);
		
		int fileId;
		int pageId;
		
		long decBufSize = strlen(dataList[blobPos]);
		
		Base64Coder b64;
		unsigned char *decBuf = (unsigned char*)malloc(decBufSize);
		if ( decBuf == NULL )
		{
		    throw Exception(EXLOC, "malloc system error");     
		}
		
		long blobSize = b64.decode(dataList[blobPos], decBuf, blobSize);
		
		_pGTM->putBlobData(tabSetId, decBuf, blobSize, fileId, pageId);
		
		free (decBuf );
		
		int len = tc.getTypeLen(pF->getType(), colVal);
		CegoField f(CegoField(pF->getTableName(), pF->getTableAlias(), colName, pF->getType(), len));
		Chain blobRef = Chain("[") + Chain(fileId) + Chain(",") + Chain(pageId) + Chain("]");
		CegoFieldValue fv(pF->getType(), blobRef);
		f.setValue(fv);
		fvl.Insert(f);
	    }
	}
	else
	{
	    // null value
	    CegoField f(CegoField(pF->getTableName(), colName));		    
	    fvl.Insert(f);
	}
	
	pF = _schema.Next();
    }
    
    
    if ( _isFirst )
    {
	_idxList.Empty();
	_keyList.Empty();
	_checkList.Empty();


	int tabSetId = _pDBMng->getTabSetId(_tableSet);	
	_pGTM->getObjectWithFix(tabSetId, tabName, CegoObject::TABLE, _oe, _bp);
	_sysEntry = CegoDataPointer(_bp.getFileId(), _bp.getPageId(), _bp.getEntryPos());

	// _pDBMng->bufferUnfix(_bp, false, _pGTM->getLockHandle());
	
	_pGTM->getObjectListByTable(_oe.getTabSetId(), _oe.getName(), _idxList, _btreeList, _keyList, _checkList);
	_isFirst = false;
    }

    CegoField* pCF = _schema.First();
    CegoField* pCV =  fvl.First();
    
    int pos=1;
    while ( pCF && pCV)
    {	    
	CegoFieldValue fv = pCV->getValue();
	CegoQueryHelper qh;
	
	qh.prepareFieldValue(pCF, fv, pos);
	
	pCV->setValue(fv); 
	
	pCF = _schema.Next();
	pCV = fvl.Next();
	
	pos++;
    }
    if ( pCF || pCV )
    {
	throw Exception(EXLOC, "Mismatched argument count for value list");
    }

    CegoDataPointer dp;
    Chain virginIndex;

    _pGTM->insertDataTable(_oe, fvl, _idxList, _btreeList, _keyList, _checkList, _sysEntry, virginIndex, dp, _doLogging);
    
}




