mariadb-columnstore-engine/dbcon/dmlpackage/insertdmlpackage.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: insertdmlpackage.cpp 9210 2013-01-21 14:10:42Z rdempsey $
 *
 *
 ***********************************************************************/
#include <stdexcept>
#include <iostream>
#include <boost/tokenizer.hpp>
#include <string>
using namespace std;
#define INSERTDMLPKG_DLLEXPORT
#include "insertdmlpackage.h"
#undef INSERTDMLPKG_DLLEXPORT

#ifdef _MSC_VER
#define strcasecmp stricmp
#endif

namespace dmlpackage
{

InsertDMLPackage::InsertDMLPackage()
{}

InsertDMLPackage::InsertDMLPackage( std::string schemaName, std::string tableName,
                                    std::string dmlStatement, int sessionID )
    : CalpontDMLPackage( schemaName, tableName, dmlStatement, sessionID )
{}

InsertDMLPackage::~InsertDMLPackage()
{}

int InsertDMLPackage::write(messageqcpp::ByteStream& bytestream)
{
    int retval = 1;

    messageqcpp::ByteStream::byte package_type = DML_INSERT;
    bytestream << package_type;

    messageqcpp::ByteStream::quadbyte session_id = fSessionID;
    bytestream << session_id;

    bytestream << fUuid;

    bytestream << fDMLStatement;
    bytestream << fDMLStatement;
    bytestream << fSchemaName;
    bytestream << fTimeZone;
    bytestream << (uint8_t)fLogging;
    bytestream << (uint8_t)fLogending;

    bytestream << fTableOid;
    bytestream << static_cast<const messageqcpp::ByteStream::byte>(fIsInsertSelect);
    bytestream << static_cast<const messageqcpp::ByteStream::byte>(fIsBatchInsert);
    bytestream << static_cast<const messageqcpp::ByteStream::byte>(fIsAutocommitOn);

    if (fTable != 0)
    {
        retval = fTable->write(bytestream);
    }

    return retval;
}

int InsertDMLPackage::read(messageqcpp::ByteStream& bytestream)
{
    int retval = 1;

    messageqcpp::ByteStream::quadbyte session_id;
    bytestream >> session_id;
    fSessionID = session_id;
    bytestream >> fUuid;

    std::string dmlStatement;
    bytestream >> fDMLStatement;
    bytestream >> fSQLStatement;
    bytestream >> fSchemaName;
    bytestream >> fTimeZone;
    uint8_t logging;
    bytestream >> logging;
    fLogging = (logging != 0);
    uint8_t logending;
    bytestream >> logending;
    fLogending = (logending != 0);

    bytestream >> fTableOid;
    bytestream >> reinterpret_cast<messageqcpp::ByteStream::byte&>(fIsInsertSelect);
    bytestream >> reinterpret_cast<messageqcpp::ByteStream::byte&>(fIsBatchInsert);
    bytestream >> reinterpret_cast<messageqcpp::ByteStream::byte&>(fIsAutocommitOn);

    fTable = new DMLTable();
    retval = fTable->read(bytestream);
    return retval;
}

void InsertDMLPackage::readMetaData(messageqcpp::ByteStream& bytestream)
{
    messageqcpp::ByteStream::quadbyte session_id;
    bytestream >> session_id;
    fSessionID = session_id;
    bytestream >> fUuid;

    std::string dmlStatement;
    bytestream >> fDMLStatement;
    bytestream >> fSQLStatement;
    bytestream >> fSchemaName;
    bytestream >> fTimeZone;
    uint8_t logging;
    bytestream >> logging;
    fLogging = (logging != 0);
    uint8_t logending;
    bytestream >> logending;
    fLogending = (logending != 0);

    bytestream >> fTableOid;
    bytestream >> reinterpret_cast<messageqcpp::ByteStream::byte&>(fIsInsertSelect);
    bytestream >> reinterpret_cast<messageqcpp::ByteStream::byte&>(fIsBatchInsert);
    bytestream >> reinterpret_cast<messageqcpp::ByteStream::byte&>(fIsAutocommitOn);

    fTable = new DMLTable();
    fTable->readMetaData(bytestream);
}

// Has to be called after InsertDMLPackage::readMetaData()
void InsertDMLPackage::readRowData(messageqcpp::ByteStream& bytestream)
{
    fTable->readRowData(bytestream);
}

int InsertDMLPackage::buildFromBuffer(std::string& buffer, int columns, int rows)
{
#ifdef DML_PACKAGE_DEBUG
    // cout << "The data buffer received: " << buffer << endl;
#endif
    int retval = 1;

    initializeTable();

    std::vector<std::string> dataList;
    typedef boost::tokenizer<boost::char_separator<char> > tokenizer;
    boost::char_separator<char> sep(",");
    tokenizer tokens(buffer, sep);

    for (tokenizer::iterator tok_iter = tokens.begin(); tok_iter != tokens.end(); ++tok_iter)
    {
        dataList.push_back(StripLeadingWhitespace(*tok_iter));

    }

    int n = 0;

    for (int i = 0; i < rows; i++)
    {
        //get a new row
        Row* aRowPtr = new Row();
        std::string colName;
        std::string colValue;

        for (int j = 0; j < columns; j++)
        {
            //Build a column list
            colName = dataList[n];
            n++;
            colValue = dataList[n];
            n++;
#ifdef DML_PACKAGE_DEBUG
            //cout << "The column data: " << colName << " " << colValue << endl;
#endif
            DMLColumn* aColumn = new DMLColumn(colName, colValue, false);
            (aRowPtr->get_ColumnList()).push_back(aColumn);
        }

        //build a row list for a table
        fTable->get_RowList().push_back(aRowPtr);
    }

    return retval;
}

int InsertDMLPackage::buildFromMysqlBuffer(ColNameList& colNameList, TableValuesMap& tableValuesMap, int columns, int rows, NullValuesBitset& nullValues )
{
    int retval = 1;

    initializeTable();
    Row* aRowPtr = new Row();
    std::string colName;
    std::vector<std::string> colValList;

    for (int j = 0; j < columns; j++)
    {
        //Build a column list
        colName = colNameList[j];

        colValList = tableValuesMap[j];

        DMLColumn* aColumn = new DMLColumn(colName, colValList, false, 0, nullValues[j]);
        (aRowPtr->get_ColumnList()).push_back(aColumn);
    }

    //build a row list for a table
    fTable->get_RowList().push_back(aRowPtr);
    aRowPtr = NULL;
    delete aRowPtr;
    return retval;
}

int InsertDMLPackage::buildFromSqlStatement(SqlStatement& sqlStatement)
{

    int retval = 1;

    InsertSqlStatement& insertStmt = dynamic_cast<InsertSqlStatement&>(sqlStatement);

    if (!insertStmt.fValuesOrQueryPtr)
        throw runtime_error("insertStmt.fValuesOrQueryPtr == NULL");

    initializeTable();
    bool isNULL = false;

    // only if we don't have a select statement
    if (0 == insertStmt.fValuesOrQueryPtr->fQuerySpecPtr)
    {

        ColumnNameList columnNameList = insertStmt.fColumnList;

        if (columnNameList.size())
        {

            ValuesList valuesList = insertStmt.fValuesOrQueryPtr->fValuesList;

            if (columnNameList.size() != valuesList.size())
            {
                throw logic_error("Column names and values count mismatch!");
            }

            Row* aRow = new Row();

            for (unsigned int i = 0; i < columnNameList.size(); i++)
            {
                DMLColumn* aColumn = new DMLColumn(columnNameList[i], valuesList[i], isNULL);
                (aRow->get_ColumnList()).push_back(aColumn);
            }

            fTable->get_RowList().push_back(aRow);

        }
        else
        {
            ValuesList valuesList = insertStmt.fValuesOrQueryPtr->fValuesList;
            ValuesList::const_iterator iter = valuesList.begin();
            Row* aRow = new Row();
            std::string colName = "";
            std::string colValue;

            while (iter != valuesList.end())
            {
                colValue = *iter;

                if ( strcasecmp(colValue.c_str(), "NULL") == 0)
                {
                    isNULL = true;
                }
                else
                {
                    isNULL = false;
                }

                DMLColumn* aColumn = new DMLColumn(colName, colValue, isNULL);
                (aRow->get_ColumnList()).push_back(aColumn);

                ++iter;
            }

            fTable->get_RowList().push_back(aRow);
        }

    }
    else
    {
        fHasFilter = true;
        fQueryString = insertStmt.getQueryString();
    }

    return retval;
}

}  // namespace dmlpackage