dbcon/ddlpackageproc/ddlindexpopulator.cpp

/* Copyright (C) 2014 InfiniDB, Inc.

   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; version 2 of
   the License.

   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; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

/***********************************************************************
 *   $Id: ddlindexpopulator.cpp 9210 2013-01-21 14:10:42Z rdempsey $
 *
 *
 ***********************************************************************/
#include <sys/types.h>
#include <sys/ipc.h>

#include "ddlindexpopulator.h"

#include "messagelog.h"
#include "dataconvert.h"
#include "joblist.h"
#include "calpontselectexecutionplan.h"
#include "distributedenginecomm.h"
#include "simplecolumn.h"
#include "resourcemanager.h"
#include "columnresult.h"

#include <boost/any.hpp>
using namespace boost;
#include <boost/algorithm/string/case_conv.hpp>
using namespace boost::algorithm;

using namespace WriteEngine;
using namespace logging;
using namespace resultset;
using namespace joblist;

using namespace std;
using namespace execplan;
using namespace ddlpackage;
using namespace messageqcpp;

namespace ddlpackageprocessor
{

bool DDLIndexPopulator::populateIndex(DDLPackageProcessor::DDLResult& result)
{
    if (makeIndexStructs() )
        insertIndex();

    result = fResult;
    return NO_ERROR != fResult.result;
}


bool DDLIndexPopulator::makeIndexStructs( )
{
    CalpontSelectExecutionPlan csep;
    makeCsep(csep);
    ResourceManager* rm;

    if (! fEC)
    {
        fEC = DistributedEngineComm::instance(rm);
        fEC->Open();
    }

    SJLP jbl = joblist::JobListFactory::makeJobList(&csep, rm);

    boost::shared_ptr<CalpontSystemCatalog> csc = CalpontSystemCatalog::makeCalpontSystemCatalog( fSessionID );
    csc->identity(CalpontSystemCatalog::EC);

    jbl->putEngineComm(fEC);
    /*
    ResultManager * result = jbl->GetResultManager();
    result->setRunning(1);
    jbl->Execute();	*/
    jbl->doQuery();

    CalpontSystemCatalog::TableName tableName;
    tableName.schema = fTable.fSchema;
    tableName.table = fTable.fName;

    CalpontSystemCatalog::OID tableOid = (csc->tableRID ( tableName )).objnum;
    CalpontSystemCatalog::NJLSysDataList sysDataList;

    for (;;)
    {
        TableBand band;
        band = jbl->projectTable(tableOid);

        if (band.getRowCount() == 0)
        {
            // No more bands, table is done
            break;
        }

        band.convertToSysDataList(sysDataList, csc);
        break;
    }

    //size_t cnt = fColNames.size();
    size_t i = 0;
    vector<ColumnResult*>::const_iterator it;
    vector<int>::const_iterator oid_iter;

    for (it = sysDataList.begin(); it != sysDataList.end(); it++)
    {
        if (isUnique())
            fUniqueColResultList.push_back(*it);

        for ( oid_iter = fOidList.begin(); oid_iter != fOidList.end(); oid_iter++ )
        {
            if ( (*it)->ColumnOID() == *oid_iter )
            {
                CalpontSystemCatalog::ColType coltype = makeIdxStruct(*it, fColNames.size(), csc);
                addColumnData(*it, coltype, i);
            }
        }

        i++;
    }

    return (fIdxValueList.size() && NO_ERROR == fResult.result );

}


void DDLIndexPopulator::makeCsep(CalpontSelectExecutionPlan&  csep)
{

    csep.sessionID(fSessionID);

    csep.txnID(fTxnID);
    csep.verID(fSessionManager->verID());

    CalpontSelectExecutionPlan::ReturnedColumnList colList;
    CalpontSelectExecutionPlan::ColumnMap colMap;
    CalpontSystemCatalog::TableColName tableColName;
    CalpontSystemCatalog::OID oid;
    tableColName.schema = fTable.fSchema;
    tableColName.table = fTable.fName;
    boost::shared_ptr<CalpontSystemCatalog> csc = CalpontSystemCatalog::makeCalpontSystemCatalog( fSessionID );
    string tableName(fTable.fSchema + "." + fTable.fName + ".");

    ColumnNameList::const_iterator cend = fColNames.end();

    for (ColumnNameList::const_iterator cname = fColNames.begin(); cname != cend; ++cname)
    {
        string fullColName(tableName + *cname);
        SRCP srcp(new SimpleColumn (fullColName, fSessionID));
        colList.push_back(srcp);
        tableColName.column = *cname;
        oid = csc->lookupOID( tableColName );
        fOidList.push_back( oid );
        colMap.insert(CalpontSelectExecutionPlan::ColumnMap::value_type(fullColName, srcp));
    }

    csep.columnMap (colMap);
    csep.returnedCols (colList);
}


CalpontSystemCatalog::ColType DDLIndexPopulator::makeIdxStruct(const ColumnResult* cr, size_t cols, boost::shared_ptr<CalpontSystemCatalog> csc )
{
    IdxStruct idx;
    idx.treeOid = fIdxOID.treeOID;
    idx.listOid = fIdxOID.listOID;
    idx.multiColFlag = cols > 1;
    CalpontSystemCatalog::ColType coltype = csc->colType(cr->ColumnOID());
    idx.idxDataType = static_cast<CalpontSystemCatalog::ColDataType>(coltype.colDataType);

    if (isDictionaryType(coltype) )
    {
        idx.idxWidth = fTOKENSIZE;
        idx.idxType = WR_CHAR;
    }//@bug 410: index sizes are either 1, 4 or 8
    else if (exeplan::isCharType(coltype))
    {
        if (1 == coltype.colWidth) idx.idxWidth = 1;
        else idx.idxWidth = (coltype.colWidth > 4) ? 8 : 4;

        idx.idxType = WR_CHAR;
    }
    else
        idx.idxWidth = coltype.colWidth;

    fIdxStructList.push_back(idx);
    return coltype;
}

void DDLIndexPopulator::addColumnData(const execplan::ColumnResult* cr, const CalpontSystemCatalog::ColType colType, int added)
{
    WriteEngine::IdxTupleList  tupleList;
    WriteEngine::IdxTuple tuple;

    for (int i = 0; i < cr->dataCount(); ++i)
    {

        WriteEngine::IdxTuple tuple ;
        convertColData( cr, i,  colType, tuple);

        if (checkConstraints( tuple, colType, i, added))
        {
            tupleList.push_back(tuple);

            if (! added )
                fRidList.push_back(cr->GetRid(i));
        }
        else
            break;
    }

    if (tupleList.size())
        fIdxValueList.push_back(tupleList);
}


void DDLIndexPopulator::convertColData(const execplan::ColumnResult* cr, int idx,  const CalpontSystemCatalog::ColType& colType, WriteEngine::IdxTuple& tuple)
{
    if (isDictionaryType(colType))
    {
        /*	tuple.data = tokenizeData ( colType, cr->GetStringData(idx) );*/
        /*	tuple.data = tokenizeData ( cr->GetRid(idx) );*/
        tuple.data = convertTokenData(cr->GetStringData(idx));
    }
    else tuple.data = convertData( colType, cr, idx);
}

boost::any DDLIndexPopulator::convertTokenData( const std::string& data )
{
    string strData((size_t)fTOKENSIZE < data.length() ? data.substr(0, fTOKENSIZE) : data);
    return  strData;
}

#if 0
// Disabled this function as it is currently not used.
// If we decide to use, we should check on the usage of fileop.getFileName().
// With iteration 17, the more common version of this getFileName() takes a
// partition and segment number in addition to an OID.  openColumnFile
// should perhaps be changed to use this updated version of getFileName().
bool DDLIndexPopulator::openColumnFile(WriteEngine::OID oid)
{
    FileOp fileOp;
    char fileName[WriteEngine::FILE_NAME_SIZE];

    if (WriteEngine::NO_ERROR == fileOp.getFileName(oid, fileName) )
    {
        fColumnFile.open(fileName);
        return true;
    }
    else
    {
        logError("Could not get column file name for data");
        return false;
    }
}
#endif

// Workaround to get original column token and not "retokenize" the string value
boost::any DDLIndexPopulator::tokenizeData( WriteEngine::RID rid )
{
    int64_t byteOffset = rid * fTOKENSIZE;
    ByteStream::byte inbuf[fTOKENSIZE];
    fColumnFile.seekg(byteOffset, ios::beg);
    fColumnFile.read(reinterpret_cast<char*>(inbuf), fTOKENSIZE);

    WriteEngine::Token token;
    memcpy(&token, inbuf, fTOKENSIZE);
    return token;
}


boost::any DDLIndexPopulator::tokenizeData( const execplan::CalpontSystemCatalog::ColType& colType, const std::string& data )
{
    WriteEngine::DctnryTuple  dictTuple;

    if ( data.length() > (unsigned int)colType.colWidth )
    {
        logError("Insert value is too large for column");
    }
    else
    {
        WriteEngine::DctnryStruct dictStruct;
        dictStruct.treeOid = colType.ddn.treeOID;
        dictStruct.listOid = colType.ddn.listOID;
        dictStruct.dctnryOid = colType.ddn.dictOID;
        dictTuple.sigValue = data.c_str();
        dictTuple.sigSize = data.length();
        int error = NO_ERROR;

        if ( NO_ERROR != (error = fWriteEngine->tokenize( fTxnID, dictStruct, dictTuple)) )
        {
            logError("Tokenization failed", error);
        }
    }

    return dictTuple.token;
}


boost::any   DDLIndexPopulator::convertData(const CalpontSystemCatalog::ColType&  colType, const execplan::ColumnResult* cr, int idx )
{
    uint64_t data = cr->GetData(idx);

    switch ( colType.colDataType )
    {
        case CalpontSystemCatalog::BIT:
        case execplan::CalpontSystemCatalog::TINYINT:
            return  *reinterpret_cast<char*>(&data);

        case execplan::CalpontSystemCatalog::SMALLINT:
            return  *reinterpret_cast<short*>(&data);

        case execplan::CalpontSystemCatalog::DATE:	// @bug 375
        case execplan::CalpontSystemCatalog::MEDINT:
        case execplan::CalpontSystemCatalog::INT:
            return  *reinterpret_cast<int*>(&data);

        case execplan::CalpontSystemCatalog::DATETIME: 	// @bug 375
        case execplan::CalpontSystemCatalog::TIME:
        case execplan::CalpontSystemCatalog::TIMESTAMP:
        case execplan::CalpontSystemCatalog::BIGINT:
            return  *reinterpret_cast<long long*>(&data);

        case execplan::CalpontSystemCatalog::DECIMAL:
        {
            if (colType.colWidth <= CalpontSystemCatalog::FOUR_BYTE) return  *reinterpret_cast<short*>(&data);

            else if (colType.colWidth <= 9)              return  *reinterpret_cast<int*>(&data);

            else                                         return  *reinterpret_cast<long long*>(&data);
        }

        case execplan::CalpontSystemCatalog::FLOAT:
            return  *reinterpret_cast<float*>(&data);

        case execplan::CalpontSystemCatalog::DOUBLE:
            return  *reinterpret_cast<double*>(&data);

        case execplan::CalpontSystemCatalog::CHAR:
        case execplan::CalpontSystemCatalog::VARCHAR:
        {
            string  strData(cr->GetStringData(idx) );
            return  *reinterpret_cast<string*>(&strData);
        }

        default:
            break;
    }

    logError("Invalid column type");
    throw std::runtime_error("Invalid data");

    return *reinterpret_cast<long long*>(&data);

}


void DDLIndexPopulator::insertIndex( )
{
// @bug 359 use bulk load build
    int rc = (1 < fIdxStructList.size()) ?
             (void)0
             :  (void)0;

    if (rc)
        logError("Error inserting index values", rc );

}

bool DDLIndexPopulator::isDictionaryType(const CalpontSystemCatalog::ColType& colType)
{
    return ( (CalpontSystemCatalog::CHAR == colType.colDataType && 8 < colType.colWidth )
             || (CalpontSystemCatalog::VARCHAR == colType.colDataType  &&  7 < colType.colWidth )
             || (CalpontSystemCatalog::DECIMAL == colType.colDataType  &&  18 < colType.precision ));

}

bool DDLIndexPopulator::checkConstraints( const IdxTuple& data, const CalpontSystemCatalog::ColType& ctype, int i, int column)
{

    switch ( fConstraint )
    {
        case DDL_INVALID_CONSTRAINT:
            return true;

        case DDL_UNIQUE:
        case DDL_PRIMARY_KEY:
            if ((size_t)column + 1 < fColNames.size() )
                return true;

            return checkUnique( i, ctype );

        case DDL_NOT_NULL:
            return checkNotNull( data, ctype );

        case DDL_CHECK:
            return checkCheck( data, ctype );

        default:
            return true; //?
    }

}

// Check if the row of data at idx is already in fUniqueColResultList

bool DDLIndexPopulator::checkUnique( int idx, const CalpontSystemCatalog::ColType& colType )
{
    if (0 == idx)
        return true;

    //Get row of data as each column result data at idx
    size_t indexSize = fColNames.size();
    vector <uint64_t> rowIntData(indexSize);
    vector <string> rowStrData(indexSize);

    for (size_t i = 0; i < indexSize; ++i)
    {
        //if  ( isStringType(fUniqueColResultList[i]->columnType()) )
        if  ( isStringType(colType.colDataType) )
            rowStrData[i] = fUniqueColResultList[i]->GetStringData(idx);
        else
            rowIntData[i] = fUniqueColResultList[i]->GetData(idx);
    }

    //check if each value in the idx row is equal to each value in a previous row
    // i is the row; j is the column.
    bool unique = true;

    for (int i = 0; i < idx && unique; ++i)
    {
        bool equal = true;

        for (size_t j = 0; j < indexSize && equal; ++j)
        {
            if ( isStringType(colType.colDataType) )
            {
                equal = fUniqueColResultList[j]->GetStringData(i) == rowStrData[j];
            }
            else
            {
                equal = (static_cast<uint64_t>(fUniqueColResultList[j]->GetData(i)) == rowIntData[j]);
            }
        }

        unique = ! equal;
    }

    if (! unique)
    {
        stringstream ss;
        ss << idx;
        logError("Unique Constraint violated on row: " + ss.str() );
    }

    return unique;
}


bool DDLIndexPopulator::checkNotNull(const IdxTuple& data, const CalpontSystemCatalog::ColType& colType)
{

    any nullvalue = DDLNullValueForType(colType);
    bool isNull = false;

    switch ( colType.colDataType )
    {
        case CalpontSystemCatalog::BIT:
            break;

        case execplan::CalpontSystemCatalog::TINYINT:
            isNull = any_cast<char>(data.data) == any_cast<char>(nullvalue);
            break;

        case execplan::CalpontSystemCatalog::SMALLINT:
            isNull = any_cast<short>(data.data) == any_cast<short>(nullvalue);
            break;

        case execplan::CalpontSystemCatalog::MEDINT:
        case execplan::CalpontSystemCatalog::INT:
            isNull = any_cast<int>(data.data) == any_cast<int>(nullvalue);
            break;

        case execplan::CalpontSystemCatalog::BIGINT:
            isNull = any_cast<long long>(data.data) == any_cast<long long>(nullvalue);
            break;

        case execplan::CalpontSystemCatalog::DECIMAL:
        {
            if (colType.colWidth <= CalpontSystemCatalog::FOUR_BYTE)
                isNull = any_cast<short>(data.data) == any_cast<short>(nullvalue);
            else if (colType.colWidth <= 9)
                isNull =  any_cast<int>(data.data) == any_cast<int>(nullvalue);
            else if (colType.colWidth <= 18)
                isNull =  any_cast<long long>(data.data) == any_cast<long long>(nullvalue);
            else
                isNull =  compareToken(any_cast<WriteEngine::Token>(data.data), any_cast<WriteEngine::Token>(nullvalue));

            break;
        }

        case execplan::CalpontSystemCatalog::FLOAT:
            isNull =  any_cast<float>(data.data) == any_cast<float>(nullvalue);
            break;

        case execplan::CalpontSystemCatalog::DOUBLE:
            isNull =  any_cast<double>(data.data) == any_cast<double>(nullvalue);
            break;

        case execplan::CalpontSystemCatalog::DATE:
            isNull =  any_cast<int>(data.data) == any_cast<int>(nullvalue);
            break;

        case execplan::CalpontSystemCatalog::DATETIME:
        case execplan::CalpontSystemCatalog::TIME:
        case execplan::CalpontSystemCatalog::TIMESTAMP:
            isNull =  any_cast<long long>(data.data) == any_cast<long long>(nullvalue);
            break;

        case execplan::CalpontSystemCatalog::CHAR:
        {
            if (colType.colWidth == execplan::CalpontSystemCatalog::ONE_BYTE)
                isNull =  any_cast<string>(data.data) == any_cast<string>(nullvalue);
            else if (colType.colWidth == execplan::CalpontSystemCatalog::TWO_BYTE)
                isNull =  any_cast<string>(data.data) == any_cast<string>(nullvalue);
            else if (colType.colWidth <= execplan::CalpontSystemCatalog::FOUR_BYTE)
                isNull =  any_cast<string>(data.data) == any_cast<string>(nullvalue);
            else
                isNull =  compareToken(any_cast<WriteEngine::Token>(data.data), any_cast<WriteEngine::Token>(nullvalue));

            break;

        }

        case execplan::CalpontSystemCatalog::VARCHAR:
        {
            if (colType.colWidth == execplan::CalpontSystemCatalog::ONE_BYTE)
                isNull =  any_cast<string>(data.data) == any_cast<string>(nullvalue);
            else if (colType.colWidth < execplan::CalpontSystemCatalog::FOUR_BYTE)
                isNull =  any_cast<string>(data.data) == any_cast<string>(nullvalue);
            else
                isNull =  compareToken(any_cast<WriteEngine::Token>(data.data), any_cast<WriteEngine::Token>(nullvalue));

            break;
        }

        default:
            throw std::runtime_error("getNullValueForType: unkown column data type");
    }

    if (isNull)
        logError("Null value not allowed in index");

    return ! isNull;

}

void  DDLIndexPopulator::logError(const string& msg, int error)
{

    Message::Args args;
    Message message(9);
    args.add((string)__FILE__ + ": ");
    args.add(msg);

    if (error)
    {
        args.add("Error number: ");
        args.add(error);
    }

    message.format( args );

    fResult.result = DDLPackageProcessor::CREATE_ERROR;
    fResult.message = message;
}


} //namespace