/* Copyright (C) 2014 InfiniDB, Inc.
   Copyright (C) 2016 MariaDB Corporation

   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. */

#include <stdio.h>
#include <string>
#include <stdexcept>
#include <sys/timeb.h>
using namespace std;

#include <boost/scoped_ptr.hpp>
using namespace boost;

#include <cppunit/extensions/HelperMacros.h>

#include "writeengine.h"
#include "we_colop.h"

using namespace WriteEngine;

extern WriteEngine::BRMWrapper* brmWrapperPtr;

class WriteEngineTest : public CppUnit::TestFixture
{
  CPPUNIT_TEST_SUITE(WriteEngineTest);

  // Column related operation testing
  /*
  CPPUNIT_TEST( testCreateColumnFiles );
  CPPUNIT_TEST( testWriteColumnFiles );
  CPPUNIT_TEST( testReadColumnFiles );

  CPPUNIT_TEST( testInsertRowInt );
  CPPUNIT_TEST( testInsertRowLonglong );
  CPPUNIT_TEST( testInsertRowDouble );
  CPPUNIT_TEST( testInsertRowChar );
  CPPUNIT_TEST( testInsertBulk );

  // Table related operation testing
  CPPUNIT_TEST( testTableAllocateRowId );

  // Wrapper interface testing
  CPPUNIT_TEST( testInterfaceCreateColumnFile );
  CPPUNIT_TEST( testInterfaceInsertUpdateDeleteRow );
  CPPUNIT_TEST( testInterfaceInsertRowMedIntDouble );

  CPPUNIT_TEST( testInterfaceInsertRowChar );

  CPPUNIT_TEST( testInterfaceInsertRowCharQAWidth4 );
  CPPUNIT_TEST( testInterfaceInsertRowCharQAWidth3 );
  CPPUNIT_TEST( testInterfaceInsertRowCharQAWidth8 );
  CPPUNIT_TEST( testInterfaceInsertRowCharQAWidth1 );


  CPPUNIT_TEST( testInterfaceInsertRowLongLong );
  CPPUNIT_TEST( testInterfaceInsertRowByte );
  CPPUNIT_TEST( testInterfaceInsertRowToken );
  CPPUNIT_TEST( testInterfaceInsertRowSingleChar );
  CPPUNIT_TEST( testInterfaceInsertRowDoubleChar );

  CPPUNIT_TEST( testInterfaceCreateIndex );
  CPPUNIT_TEST( testInterfaceDropIndex );
  CPPUNIT_TEST( testInterfaceUpdateIndexChar );
  CPPUNIT_TEST( testInterfaceUpdateMultiColIndex);

  CPPUNIT_TEST( testInterfaceDctnryToken );
  CPPUNIT_TEST( testInterfaceDctnryTokenRollBack );

  CPPUNIT_TEST( testInsertCommitRollback );
  CPPUNIT_TEST( testInterfaceInsertRowHwm );

  CPPUNIT_TEST( testInterfaceCombineIndexLoad );   // note: this test case must be the last one
  //CPPUNIT_TEST( testTmpBulkPerformance );   // note: this test case must be the last one
  */
  // CPPUNIT_TEST(testBoostFloat);
  /*
  CPPUNIT_TEST(testCreateColumnFiles1);
  CPPUNIT_TEST(testCreateColumnFiles2);
  CPPUNIT_TEST(testCreateColumnFiles3);
  CPPUNIT_TEST(testCreateColumnFiles4);
  */

  // CPPUNIT_TEST(testCreateFileMultipleIONBF);
  // CPPUNIT_TEST(testCreateFile64MBIONBF);
  // CPPUNIT_TEST(testCreateFileMultipleBF);
  // CPPUNIT_TEST(testCreateFilelargerBF);
  CPPUNIT_TEST(testSoloman);
  /*
  CPPUNIT_TEST(testCreateFileNoBRM);
  CPPUNIT_TEST(testCreateFileBRM);
  */
  CPPUNIT_TEST_SUITE_END();

 private:
  WriteEngineWrapper m_wrapper;
  Session m_session;

 public:
  void setUp()
  {
    m_wrapper.setDebugLevel(DEBUG_3);
    m_session.txnid = 10;
    BRMWrapper::setUseBrm(true);
  }

  void tearDown()
  {
  }

  void SetColumnStruct(ColStruct& column, OID dataOid, int colWidth,
                       CalpontSystemCatalog::ColDataType colDataType)
  {
    column.dataOid = dataOid;
    column.colWidth = colWidth;
    column.colDataType = colDataType;
    column.tokenFlag = false;
  }

  void CreateColumnFile(FID fid, int width, CalpontSystemCatalog::ColDataType colDataType, ColType colType)
  {
    int rc;
    ColumnOp colOp;
    Column curCol;

    colOp.deleteFile(fid);
    CPPUNIT_ASSERT(colOp.exists(fid) == false);

    // create column files
    rc = colOp.createColumn(curCol, 2, width, colDataType, colType, fid);
    CPPUNIT_ASSERT(rc == NO_ERROR);
  }

  void testCreateColumnFiles()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;

    colOp.initColumn(curCol);

    // file opertaions
    colOp.deleteFile(100);
    CPPUNIT_ASSERT(colOp.exists(100) == false);

    colOp.deleteFile(101);
    CPPUNIT_ASSERT(colOp.exists(101) == false);

    colOp.deleteFile(103);
    CPPUNIT_ASSERT(colOp.exists(103) == false);

    rc = colOp.createColumn(curCol, 2, 50, WriteEngine::CHAR, WriteEngine::WR_CHAR, 101);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    colOp.deleteFile(101);
    rc = colOp.createColumn(curCol, 2, 3, WriteEngine::CHAR, WriteEngine::WR_CHAR, 101);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    colOp.deleteFile(101);
    rc = colOp.createColumn(curCol, 2, 5, WriteEngine::CHAR, WriteEngine::WR_CHAR, 101);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    // create column files
    rc = colOp.createColumn(curCol, 2, 8, WriteEngine::CHAR, WriteEngine::WR_CHAR, 100);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = colOp.createColumn(curCol, 2, 8, WriteEngine::CHAR, WriteEngine::WR_CHAR, 100);
    CPPUNIT_ASSERT(rc == ERR_FILE_EXIST);

    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    colOp.clearColumn(curCol);

    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == ERR_INVALID_PARAM);

    rc = colOp.createTable();
    CPPUNIT_ASSERT(rc == NO_ERROR);
  }

  void testWriteColumnFiles()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;
    unsigned char buf[BYTE_PER_BLOCK];

    colOp.initColumn(curCol);
    CPPUNIT_ASSERT(curCol.colNo == 0);
    CPPUNIT_ASSERT(curCol.colWidth == 0);
    CPPUNIT_ASSERT(curCol.dataFile.pFile == NULL);
    //      CPPUNIT_ASSERT( curCol.bitmapFile.pFile == NULL );
    CPPUNIT_ASSERT(curCol.dataFile.fid == 0);
    //      CPPUNIT_ASSERT( curCol.bitmapFile.fid == 0 );

    colOp.setColParam(curCol, 2, 4, INT, WriteEngine::WR_INT, 100);
    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    int testVal = 60;
    memset(buf, 0, BYTE_PER_BLOCK);
    memcpy(buf + 8, &testVal, 4);

    rc = colOp.writeDBFileFbo(curCol.dataFile.pFile, buf, 1, 1);  // block 2
    CPPUNIT_ASSERT(rc == NO_ERROR);

    colOp.clearColumn(curCol);

    CPPUNIT_ASSERT(curCol.colNo == 0);
    CPPUNIT_ASSERT(curCol.colWidth == 0);
    CPPUNIT_ASSERT(curCol.dataFile.pFile == NULL);
    //      CPPUNIT_ASSERT( curCol.bitmapFile.pFile == NULL );
    CPPUNIT_ASSERT(curCol.dataFile.fid == 0);
    //      CPPUNIT_ASSERT( curCol.bitmapFile.fid == 0 );

    CPPUNIT_ASSERT(colOp.isValid(curCol) == false);
  }

  void testReadColumnFiles()
  {
    ColumnOp colOp;
    Column curCol, errCol;
    int rc;
    unsigned char buf[BYTE_PER_BLOCK];
    CalpontSystemCatalog::ColDataType colDataType;

    //      colOp.setUseBrm(false);
    CPPUNIT_ASSERT(colOp.getColDataType("integer", colDataType) == true);
    CPPUNIT_ASSERT(colDataType == INT);
    CPPUNIT_ASSERT(colOp.getColDataType("Int1", colDataType) == false);

    // check error situation
    colOp.initColumn(errCol);
    colOp.setColParam(errCol, 2, 8, CHAR, WriteEngine::WR_CHAR, 103);
    rc = colOp.openColumnFile(errCol);
    CPPUNIT_ASSERT(rc == ERR_FILE_READ);
    colOp.clearColumn(errCol);

    colOp.initColumn(errCol);
    colOp.setColParam(errCol, 2, 8, CHAR, WriteEngine::WR_CHAR, 103);
    rc = colOp.openColumnFile(errCol);
    CPPUNIT_ASSERT(rc == ERR_FILE_READ);
    colOp.clearColumn(errCol);

    // check normal situation
    colOp.initColumn(curCol);

    colOp.setColParam(curCol, 2, 8, CHAR, WriteEngine::WR_CHAR, 100);
    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    memset(buf, 0, BYTE_PER_BLOCK);

    rc = colOp.readDBFile(curCol.dataFile.pFile, buf, 1, true);  // block 2
    CPPUNIT_ASSERT(rc == NO_ERROR);

    int testVal = 0;
    memcpy(&testVal, buf + 8, 8);
    printf("\nread the test value : %d", testVal);
    CPPUNIT_ASSERT(testVal == 60);

    colOp.clearColumn(curCol);
  }

  void testInsertRowInt()
  {
    ColumnOp colOp;
    Column curCol;
    int rc, valArray[5], oldValArray[5];
    RID rowIdArray[5];

    rowIdArray[0] = 1;
    rowIdArray[1] = 3;
    rowIdArray[2] = 4;
    rowIdArray[3] = 7;
    rowIdArray[4] = 8;

    valArray[0] = 8;
    valArray[1] = 5;
    valArray[2] = 3;
    valArray[3] = 0;
    valArray[4] = 16;

    CreateColumnFile(100, 4, WriteEngine::INT, WriteEngine::WR_INT);

    colOp.initColumn(curCol);
    colOp.setColParam(curCol, 2, 4, INT, WriteEngine::WR_INT, 100);
    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = colOp.writeRow(curCol, 5, (RID*)rowIdArray, valArray, oldValArray);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    for (int i = 0; i < 5; i++)
      CPPUNIT_ASSERT(oldValArray[i] == (int)0x80000001);

    colOp.clearColumn(curCol);
  }

  void testInsertRowLonglong()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;
    RID rowIdArray[3];
    int width = 8, totalRow = 3;
    FID fid = 100;
    long long valArray[3];
    uint64_t oldValArray[3], verifyArray[3];
    CalpontSystemCatalog::ColDataType colDataType = CalpontSystemCatalog::BIGINT;
    ColType colType = WriteEngine::WR_LONGLONG;
    uint64_t emptyVal = 0x8000000000000001LL;

    rowIdArray[0] = 1;
    rowIdArray[1] = 3;
    rowIdArray[2] = 4;

    verifyArray[0] = valArray[0] = 32111238;
    verifyArray[1] = valArray[1] = 1231235;
    verifyArray[2] = valArray[2] = 67731233;

    CreateColumnFile(fid, width, colDataType, colType);

    colOp.initColumn(curCol);
    colOp.setColParam(curCol, 2, width, colDataType, colType, fid);
    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = colOp.writeRow(curCol, totalRow, (RID*)rowIdArray, valArray, oldValArray);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    for (int i = 0; i < totalRow; i++)
      CPPUNIT_ASSERT(oldValArray[i] == emptyVal);

    valArray[0] = 1900003;
    valArray[1] = 2349000;
    valArray[2] = 78900123;

    rc = colOp.writeRow(curCol, totalRow, (RID*)rowIdArray, valArray, oldValArray);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    for (int i = 0; i < totalRow; i++)
      CPPUNIT_ASSERT(oldValArray[i] == verifyArray[i]);

    colOp.clearColumn(curCol);
  }

  void testInsertRowDouble()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;
    RID rowIdArray[3];
    int width = 8, totalRow = 3;
    FID fid = 100;
    double valArray[3];
    int64_t oldValArray[3];
    CalpontSystemCatalog::ColDataType colDataType = CalpontSystemCatalog::DOUBLE;
    ColType colType = WriteEngine::WR_DOUBLE;
    int64_t emptyVal = 0xFFFAAAAAAAAAAAABLL;

    rowIdArray[0] = 1;
    rowIdArray[1] = 3;
    rowIdArray[2] = 4;

    valArray[0] = 8;
    valArray[1] = 5;
    valArray[2] = 3;

    CreateColumnFile(fid, width, colDataType, colType);

    colOp.initColumn(curCol);
    colOp.setColParam(curCol, 2, width, colDataType, colType, fid);
    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = colOp.writeRow(curCol, totalRow, (RID*)rowIdArray, valArray, oldValArray);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    for (int i = 0; i < totalRow; i++)
      CPPUNIT_ASSERT(oldValArray[i] == emptyVal);

    colOp.clearColumn(curCol);
  }

  void testInsertRowChar()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;
    RID rowIdArray[3];
    int width = 8, totalRow = 3;
    FID fid = 100;
    char valArray[24], buf[8];
    char testValue[3][8] = {"abc", "eDFFF", "GHK"};
    uint64_t oldValArray[3];
    CalpontSystemCatalog::ColDataType colDataType = CalpontSystemCatalog::CHAR;
    ColType colType = WriteEngine::WR_CHAR;
    uint64_t emptyVal = 0xFFFFFFFFFFFFFFFFLL;

    rowIdArray[0] = 1;
    rowIdArray[1] = 3;
    rowIdArray[2] = 4;

    memset(valArray, 0, 24);
    colOp.writeBufValue((unsigned char*)buf, testValue[0], 8, true);
    memcpy(valArray, buf, 8);
    colOp.writeBufValue((unsigned char*)buf, testValue[1], 8, true);
    memcpy(valArray + 8, buf, 8);
    colOp.writeBufValue((unsigned char*)buf, testValue[2], 8, true);
    memcpy(valArray + 16, buf, 8);

    CreateColumnFile(fid, width, colDataType, colType);

    colOp.initColumn(curCol);
    colOp.setColParam(curCol, 2, width, colDataType, colType, fid);
    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = colOp.writeRow(curCol, totalRow, (RID*)rowIdArray, valArray, oldValArray);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    for (int i = 0; i < totalRow; i++)
      CPPUNIT_ASSERT(oldValArray[i] == emptyVal);

    colOp.clearColumn(curCol);
  }

  void testInsertBulk()
  {
    FILE* pSourceFile = NULL;
    ColumnOp colOp;
    Column curCol;
    int rc, width = 8;
    int hwm;
    FID fid = 999;
    CalpontSystemCatalog::ColDataType colDataType = CalpontSystemCatalog::CHAR;
    ColType colType = WriteEngine::WR_CHAR;

    colOp.deleteFile(999);
    colOp.deleteFile(998);
    CPPUNIT_ASSERT(colOp.exists(999) == false);
    CPPUNIT_ASSERT(colOp.exists(998) == false);

    int allocSize = 0;
    CPPUNIT_ASSERT(colOp.createFile(998, 20, allocSize, 0xEEEE, 2) == NO_ERROR);
    CreateColumnFile(fid, width, colDataType, colType);

    colOp.initColumn(curCol);
    colOp.setColParam(curCol, 2, width, colDataType, colType, fid);
    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    pSourceFile = colOp.openFile(998);
    CPPUNIT_ASSERT(pSourceFile != NULL);

    hwm = colOp.getFileSize(curCol.dataFile.pFile) / BYTE_PER_BLOCK - 5;
    rc = colOp.insertBulk(curCol, pSourceFile, hwm, 20);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    colOp.clearColumn(curCol);
    colOp.closeFile(pSourceFile);
  }

  void testTableAllocateRowId()
  {
    ColumnOp colOp;
    Column curCol;
    RID rowIdArray[3];
    int rc;
    int width = 4, hwm = 0;
    FID fid = 100;
    CalpontSystemCatalog::ColDataType colDataType = CalpontSystemCatalog::INT;
    ColType colType = WriteEngine::WR_INT;

    CreateColumnFile(fid, width, colDataType, colType);

    colOp.initColumn(curCol);
    colOp.setColParam(curCol, 2, width, colDataType, colType, fid);

    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    colOp.allocRowId(curCol, 3, rowIdArray, hwm);
    CPPUNIT_ASSERT(rowIdArray[0] == 0);
    CPPUNIT_ASSERT(rowIdArray[1] == 1);
    CPPUNIT_ASSERT(rowIdArray[2] == 2);
  }

  void testInterfaceCreateColumnFile()
  {
    ColumnOp colOp;
    FID fid = 100;

    if (colOp.exists(fid))
    {
      CPPUNIT_ASSERT(m_wrapper.dropColumn(m_session.txnid, fid) == NO_ERROR);
      CPPUNIT_ASSERT(m_wrapper.dropColumn(m_session.txnid, fid) == ERR_FILE_NOT_EXIST);
    }

    CPPUNIT_ASSERT(m_wrapper.createColumn(m_session.txnid, fid, WriteEngine::DOUBLE, 8) == NO_ERROR);
    CPPUNIT_ASSERT(colOp.exists(100) == true);
  }

  void testInterfaceInsertUpdateDeleteRow()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    int rc, total, valArray[3];
    float dVal;
    //      ColumnOp                colOp;

    //      colOp = m_wrapper.getColumnOp();

    CreateColumnFile(100, 4, WriteEngine::INT, WriteEngine::WR_INT);
    CreateColumnFile(200, 4, WriteEngine::FLOAT, WriteEngine::WR_FLOAT);

    // test column struct list
    SetColumnStruct(curColStruct, 100, 4, WriteEngine::INT);
    colStructList.push_back(curColStruct);

    SetColumnStruct(curColStruct, 200, 4, WriteEngine::FLOAT);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == 4);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::INT);

    testColStruct = colStructList[1];
    CPPUNIT_ASSERT(testColStruct.dataOid == 200);
    CPPUNIT_ASSERT(testColStruct.colWidth == 4);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::FLOAT);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // test column values
    // add values for the first column
    curTuple.data = 3;
    curTupleList.push_back(curTuple);

    curTuple.data = 0;
    curTupleList.push_back(curTuple);

    curTuple.data = 100;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    m_wrapper.convertValArray(3, WriteEngine::WR_INT, curTupleList, valArray);
    CPPUNIT_ASSERT(valArray[0] == 3);
    CPPUNIT_ASSERT(valArray[1] == 0);
    CPPUNIT_ASSERT(valArray[2] == 100);

    curTupleList.clear();

    // add values for the second column
    curTuple.data = 0.0f;
    curTupleList.push_back(curTuple);

    curTuple.data = 1234.78f;
    curTupleList.push_back(curTuple);

    curTuple.data = 999.98f;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    float valArray1[3];
    m_wrapper.convertValArray(3, WriteEngine::WR_FLOAT, curTupleList, valArray1);
    CPPUNIT_ASSERT(valArray1[0] == 0.0f);
    CPPUNIT_ASSERT(valArray1[1] == 1234.78f);
    CPPUNIT_ASSERT(valArray1[2] == 999.98f);

    // retrieve the values back
    // first column
    testTupleList = static_cast<ColTupleList>(colValueList[0]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 3);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 0);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 100);

    // second column
    testTupleList = static_cast<ColTupleList>(colValueList[1]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    dVal = boost::any_cast<float>(testTuple.data);
    CPPUNIT_ASSERT(dVal == 0.0f);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<float>(testTuple.data) == 1234.78f);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<float>(testTuple.data) == 999.98f);

    //      m_wrapper.printInputValue( colStructList, colValueList, ridList );

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 3);
    CPPUNIT_ASSERT(ridList[1] == 4);
    CPPUNIT_ASSERT(ridList[2] == 5);

    // try to update rows
    colValueList.clear();
    curTupleList.clear();

    curTuple.data = 9;
    curTupleList.push_back(curTuple);

    curTuple.data = 15;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);

    // add values for the second column
    curTupleList.clear();

    curTuple.data = 1.99f;
    curTupleList.push_back(curTuple);

    curTuple.data = 3000.15f;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);

    ridList.clear();
    ridList.push_back((RID)2);
    ridList.push_back((RID)3);

    rc = m_wrapper.updateColumnRec(m_session.txnid, colStructList, colValueList, colOldValList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(colOldValList.size() == 2);

    m_wrapper.printInputValue(colStructList, colOldValList, ridList);
    curTupleList = colOldValList[0];
    curTuple = curTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<int>(curTuple.data) == 100);
    curTuple = curTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<int>(curTuple.data) == 3);

    curTupleList = colOldValList[1];
    curTuple = curTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<float>(curTuple.data) == 999.98f);
    curTuple = curTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<float>(curTuple.data) == 0.0f);

    ridList[0] = 1;
    ridList[1] = 2;

    rc = m_wrapper.deleteRow(m_session.txnid, colStructList, colOldValList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(colOldValList.size() == 2);

    m_wrapper.printInputValue(colStructList, colOldValList, ridList);
    curTupleList = colOldValList[0];
    curTuple = curTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<int>(curTuple.data) == 0);
    curTuple = curTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<int>(curTuple.data) == 9);

    curTupleList = colOldValList[1];
    curTuple = curTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<float>(curTuple.data) == 1234.78f);
    curTuple = curTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<float>(curTuple.data) == 1.99f);
  }

  void testInterfaceInsertRowMedIntDouble()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    int total, rc;
    double dVal;

    CreateColumnFile(100, 4, WriteEngine::MEDINT, WriteEngine::WR_INT);
    CreateColumnFile(200, 8, WriteEngine::DOUBLE, WriteEngine::WR_DOUBLE);

    // test column struct list
    SetColumnStruct(curColStruct, 100, 4, WriteEngine::MEDINT);
    colStructList.push_back(curColStruct);

    SetColumnStruct(curColStruct, 200, 8, WriteEngine::DOUBLE);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == 4);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::MEDINT);

    testColStruct = colStructList[1];
    CPPUNIT_ASSERT(testColStruct.dataOid == 200);
    CPPUNIT_ASSERT(testColStruct.colWidth == 8);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::DOUBLE);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // test column values
    // add values for the first column
    curTuple.data = 102;
    curTupleList.push_back(curTuple);

    curTuple.data = 0;
    curTupleList.push_back(curTuple);

    curTuple.data = 200;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    // add values for the second column
    curTuple.data = 0.0;
    curTupleList.push_back(curTuple);

    curTuple.data = 1234.78;
    curTupleList.push_back(curTuple);

    curTuple.data = 999.98;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // add one rowId
    ridList.push_back((RID)5);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_ROWID_VALUE_NOT_MATCH);
    ridList.push_back((RID)9);
    ridList.push_back((RID)10);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<ColTupleList>(colValueList[0]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 102);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 0);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 200);

    // second column
    testTupleList = static_cast<ColTupleList>(colValueList[1]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    dVal = boost::any_cast<double>(testTuple.data);
    CPPUNIT_ASSERT(dVal == 0.0);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<double>(testTuple.data) == 1234.78);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<double>(testTuple.data) == 999.98);

    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );

    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 3);
    CPPUNIT_ASSERT(ridList[1] == 4);
    CPPUNIT_ASSERT(ridList[2] == 5);
  }

  void testInterfaceInsertRowChar()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    std::string curStr, testStr;
    int total, rc, width = 5;

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc != NO_ERROR);

    rc = m_wrapper.updateColumnRec(m_session.txnid, colStructList, colValueList, colOldValList, ridList);
    CPPUNIT_ASSERT(rc != NO_ERROR);

    // test column struct list
    CreateColumnFile(100, width, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 100, width, WriteEngine::CHAR);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == width);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // test column values
    // add values for the first column
    curStr = "aaaaa";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "bbbbb";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "ccccc";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // add one rowId
    ridList.push_back((RID)5);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_ROWID_VALUE_NOT_MATCH);
    ridList.push_back((RID)9);
    ridList.push_back((RID)10);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<ColTupleList>(colValueList[0]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "aaaaa");

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "bbbbb");

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "ccccc");

    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 3);
    CPPUNIT_ASSERT(ridList[1] == 4);
    CPPUNIT_ASSERT(ridList[2] == 5);
  }

  void testInterfaceInsertRowCharQAWidth4()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    std::string curStr, testStr;
    int rc, width = 4;

    // test column struct list
    CreateColumnFile(100, width, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 100, width, WriteEngine::CHAR);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == width);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    // test column values
    // add values for the first column
    curStr = "aaaa";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "aaa";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "aa";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "a";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);
    CPPUNIT_ASSERT(ridList[3] == 3);

    // separate inserts
    colValueList.clear();
    curStr = "bbbb";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 4);

    colValueList.clear();
    curStr = "bbb";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 5);

    colValueList.clear();
    curStr = "bb";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 6);

    colValueList.clear();
    curStr = "b";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 7);
  }

  void testInterfaceInsertRowCharQAWidth3()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    std::string curStr, testStr;
    int rc, width = 3;

    // test column struct list
    CreateColumnFile(100, width, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 100, width, WriteEngine::CHAR);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == width);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    // test column values
    // add values for the first column
    curStr = "aaa";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "aa";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "a";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // separate inserts
    colValueList.clear();
    curStr = "bbb";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 3);

    colValueList.clear();
    curStr = "bb";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 4);

    colValueList.clear();
    curStr = "b";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 5);
  }

  void testInterfaceInsertRowCharQAWidth8()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    std::string curStr, testStr;
    int rc, width = 8;

    // test column struct list
    CreateColumnFile(100, width, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 100, width, WriteEngine::CHAR);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == width);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    // test column values
    // add values for the first column
    curStr = "12345678";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "123456";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "1234";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // separate inserts
    colValueList.clear();
    curStr = "bbbb1234";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 3);

    colValueList.clear();
    curStr = "bb";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 4);

    colValueList.clear();
    curStr = "b";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 5);
  }

  void testInterfaceInsertRowCharQAWidth1()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    std::string curStr, testStr;
    int rc, width = 1;

    // test column struct list
    CreateColumnFile(100, width, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 100, width, WriteEngine::CHAR);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == width);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    // test column values
    // add values for the first column
    curStr = "a";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "b";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    curStr = "c";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // separate inserts
    colValueList.clear();
    curStr = "1";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 3);

    colValueList.clear();
    curStr = "2";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 4);

    colValueList.clear();
    curStr = "3";
    curTuple.data = curStr;
    curTupleList.push_back(curTuple);
    colValueList.push_back(curTupleList);
    curTupleList.clear();
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 5);
  }

  void testInterfaceInsertRowSingleChar()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    string curString;
    int total, rc;
    ColumnOp colOp;

    //      colOp = m_wrapper.getColumnOp();

    colOp.deleteFile(100);

    // test column struct list
    CreateColumnFile(100, 1, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 100, 1, WriteEngine::CHAR);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == 1);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    // test column values
    // add values for the first column
    curString = "\376";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "a";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "c";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "G";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<ColTupleList>(colValueList[0]);
    total = testTupleList.size();
    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "a");

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "c");

    testTuple = testTupleList[3];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "G");

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    printf("\nrc=%d", rc);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);
    CPPUNIT_ASSERT(ridList[3] == 3);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 4);
    CPPUNIT_ASSERT(ridList[1] == 5);
    CPPUNIT_ASSERT(ridList[2] == 6);
    CPPUNIT_ASSERT(ridList[3] == 7);
  }

  void testInterfaceInsertRowDoubleChar()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    string curString;
    int total, rc;
    ColumnOp colOp;

    //      colOp = m_wrapper.getColumnOp();

    colOp.deleteFile(100);

    // test column struct list
    CreateColumnFile(100, 2, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 100, 2, WriteEngine::CHAR);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == 2);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    // test column values
    // add values for the first column
    curString = "\377\376";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "ab";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "c";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "Ge";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<ColTupleList>(colValueList[0]);
    total = testTupleList.size();
    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "ab");

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "c");

    testTuple = testTupleList[3];
    CPPUNIT_ASSERT(boost::any_cast<std::string>(testTuple.data) == "Ge");

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    printf("\nrc=%d", rc);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);
    CPPUNIT_ASSERT(ridList[3] == 3);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 4);
    CPPUNIT_ASSERT(ridList[1] == 5);
    CPPUNIT_ASSERT(ridList[2] == 6);
    CPPUNIT_ASSERT(ridList[3] == 7);
  }

  void testInterfaceInsertRowLongLong()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    int total, rc;

    CreateColumnFile(100, 8, WriteEngine::BIGINT, WriteEngine::WR_LONGLONG);
    CreateColumnFile(200, 2, WriteEngine::SMALLINT, WriteEngine::WR_SHORT);

    // test column struct list
    SetColumnStruct(curColStruct, 100, 8, WriteEngine::BIGINT);
    colStructList.push_back(curColStruct);

    SetColumnStruct(curColStruct, 200, 2, WriteEngine::SMALLINT);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == 8);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::BIGINT);

    testColStruct = colStructList[1];
    CPPUNIT_ASSERT(testColStruct.dataOid == 200);
    CPPUNIT_ASSERT(testColStruct.colWidth == 2);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::SMALLINT);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // test column values
    // add values for the first column
    curTuple.data = (long long)1021231;
    curTupleList.push_back(curTuple);

    curTuple.data = (long long)0;
    curTupleList.push_back(curTuple);

    curTuple.data = (long long)93232200;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    // add values for the second column
    curTuple.data = (short)1000;
    curTupleList.push_back(curTuple);

    curTuple.data = (short)5678;
    curTupleList.push_back(curTuple);

    curTuple.data = (short)9;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // add one rowId
    ridList.push_back((RID)5);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_ROWID_VALUE_NOT_MATCH);
    ridList.push_back((RID)9);
    ridList.push_back((RID)10);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<ColTupleList>(colValueList[0]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<long long>(testTuple.data) == 1021231);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<long long>(testTuple.data) == 0);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<long long>(testTuple.data) == 93232200);

    // second column
    testTupleList = static_cast<ColTupleList>(colValueList[1]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<short>(testTuple.data) == 1000);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<short>(testTuple.data) == 5678);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<short>(testTuple.data) == 9);

    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );

    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 3);
    CPPUNIT_ASSERT(ridList[1] == 4);
    CPPUNIT_ASSERT(ridList[2] == 5);
  }

  void testInterfaceInsertRowByte()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    int total, rc;
    ColumnOp colOp;

    CreateColumnFile(100, 1, WriteEngine::TINYINT, WriteEngine::WR_BYTE);
    CreateColumnFile(200, 4, WriteEngine::INT, WriteEngine::WR_INT);

    // test column struct list
    SetColumnStruct(curColStruct, 100, 1, WriteEngine::TINYINT);
    colStructList.push_back(curColStruct);

    SetColumnStruct(curColStruct, 200, 4, WriteEngine::INT);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == 1);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::TINYINT);

    testColStruct = colStructList[1];
    CPPUNIT_ASSERT(testColStruct.dataOid == 200);
    CPPUNIT_ASSERT(testColStruct.colWidth == 4);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::INT);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // test column values
    // add values for the first column
    curTuple.data = (char)21;
    curTupleList.push_back(curTuple);

    curTuple.data = (char)-30;
    curTupleList.push_back(curTuple);

    curTuple.data = (char)127;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    // add values for the second column
    curTuple.data = 1000;
    curTupleList.push_back(curTuple);

    curTuple.data = 5678;
    curTupleList.push_back(curTuple);

    curTuple.data = 9;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // add one rowId
    ridList.push_back((RID)5);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_ROWID_VALUE_NOT_MATCH);
    ridList.push_back((RID)9);
    ridList.push_back((RID)10);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<ColTupleList>(colValueList[0]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<char>(testTuple.data) == 21);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<char>(testTuple.data) == -30);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<char>(testTuple.data) == 127);

    // second column
    testTupleList = static_cast<ColTupleList>(colValueList[1]);
    total = testTupleList.size();
    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 1000);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 5678);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 9);

    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );

    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 3);
    CPPUNIT_ASSERT(ridList[1] == 4);
    CPPUNIT_ASSERT(ridList[2] == 5);
  }

  void testInterfaceInsertRowToken()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    std::string curStr, testStr;
    int rc;
    ColumnOp colOp;
    OID dctnryOID = 800, treeOID = 801, listOID = 802;
    DctnryTuple dctnryTuple;

    // no matter what happened, drop stores
    //      BRMWrapper::setUseBrm(false);
    m_wrapper.dropStore(m_session.txnid, dctnryOID, treeOID, listOID);
    rc = m_wrapper.createStore(m_session.txnid, dctnryOID, treeOID, listOID);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    rc = m_wrapper.openStore(m_session.txnid, dctnryOID, treeOID, listOID);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    // colOp = m_wrapper.getColumnOp();

    // test column struct list
    CreateColumnFile(100, 8, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 100, 8, WriteEngine::CHAR);
    curColStruct.tokenFlag = true;
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 100);
    CPPUNIT_ASSERT(testColStruct.colWidth == 8);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // test column values
    // add values for the first column
    strcpy((char*)dctnryTuple.sigValue, "ABCD 12345");
    dctnryTuple.sigSize = 10;
    rc = m_wrapper.tokenize(m_session.txnid, dctnryTuple);

    if (rc != NO_ERROR)
      printf("1197 Tokenize failed, rc %i\n", rc);

    CPPUNIT_ASSERT(rc == NO_ERROR);
    curTuple.data = dctnryTuple.token;
    curTupleList.push_back(curTuple);

    Token test1;
    test1 = boost::any_cast<Token>(curTuple.data);
    //      printf( "\ncurTuple.datatype = %d\n", curTuple.data.type() );

    strcpy((char*)dctnryTuple.sigValue, "CBED 1334678");
    dctnryTuple.sigSize = 12;
    rc = m_wrapper.tokenize(m_session.txnid, dctnryTuple);

    if (rc != NO_ERROR)
      printf("1208 Tokenize failed, rc %i\n", rc);

    CPPUNIT_ASSERT(rc == NO_ERROR);
    curTuple.data = dctnryTuple.token;
    curTupleList.push_back(curTuple);

    strcpy((char*)dctnryTuple.sigValue, "GHED 2334");
    dctnryTuple.sigSize = 9;
    rc = m_wrapper.tokenize(m_session.txnid, dctnryTuple);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    curTuple.data = dctnryTuple.token;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    // add one rowId
    ridList.push_back((RID)5);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_ROWID_VALUE_NOT_MATCH);
    ridList.push_back((RID)9);
    ridList.push_back((RID)10);
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
    CPPUNIT_ASSERT(m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList) ==
                   NO_ERROR);
    CPPUNIT_ASSERT(ridList[0] == 0);
    CPPUNIT_ASSERT(ridList[1] == 1);
    CPPUNIT_ASSERT(ridList[2] == 2);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    CPPUNIT_ASSERT(ridList[0] == 3);
    CPPUNIT_ASSERT(ridList[1] == 4);
    CPPUNIT_ASSERT(ridList[2] == 5);

    m_wrapper.dropStore(m_session.txnid, dctnryOID, treeOID, listOID);
  }

  void testInterfaceCreateIndex()
  {
    int rc;
    ColumnOp colOp;

    if (colOp.exists(900))
    {
      rc = m_wrapper.dropIndex(m_session.txnid, 900, 901);
      CPPUNIT_ASSERT(rc == NO_ERROR);
    }

    rc = m_wrapper.createIndex(m_session.txnid, 900, 901);
    CPPUNIT_ASSERT(rc == NO_ERROR);
  }

  void testInterfaceDropIndex()
  {
    int rc;

    rc = m_wrapper.dropIndex(m_session.txnid, 900, 901);
    CPPUNIT_ASSERT(rc == NO_ERROR);
  }

  void SetIndexStruct(IdxStruct& curStruct, OID treeOid, OID listOid, int width,
                      CalpontSystemCatalog::ColDataType dataType)
  {
    curStruct.treeOid = treeOid;
    curStruct.listOid = listOid;
    curStruct.idxWidth = width;
    curStruct.idxDataType = dataType;
    curStruct.tokenFlag = false;
  }

  void testInterfaceUpdateIndexChar()
  {
    IdxStruct curStruct, testStruct;
    IdxTuple curTuple, testTuple;
    IdxTupleList curTupleList, testTupleList;
    IdxStructList idxStructList;
    IdxValueList idxValueList;
    RIDList ridList;
    std::string curStr, testStr;
    int total, rc;

    testInterfaceCreateIndex();

    CPPUNIT_ASSERT(m_wrapper.updateIndexRec(m_session.txnid, idxStructList, idxValueList, ridList) !=
                   NO_ERROR);

    // test column struct list
    SetIndexStruct(curStruct, 900, 901, 4, WriteEngine::MEDINT);
    idxStructList.push_back(curStruct);

    testStruct = idxStructList[0];
    CPPUNIT_ASSERT(testStruct.treeOid == 900);
    CPPUNIT_ASSERT(testStruct.listOid == 901);
    CPPUNIT_ASSERT(testStruct.idxWidth == 4);
    CPPUNIT_ASSERT(testStruct.idxDataType == WriteEngine::MEDINT);

    CPPUNIT_ASSERT(m_wrapper.checkIndexValid(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // add values for the first index
    curTuple.data = 102;
    curTupleList.push_back(curTuple);

    curTuple.data = 102;
    curTupleList.push_back(curTuple);

    curTuple.data = 102;
    curTupleList.push_back(curTuple);

    curTuple.data = 0;
    curTupleList.push_back(curTuple);

    curTuple.data = 200;
    curTupleList.push_back(curTuple);

    idxValueList.push_back(curTupleList);

    curTupleList.clear();

    // add one rowId
    ridList.push_back((RID)5);
    CPPUNIT_ASSERT(m_wrapper.checkIndexValid(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   ERR_ROWID_VALUE_NOT_MATCH);
    ridList.push_back((RID)15);
    ridList.push_back((RID)16);
    ridList.push_back((RID)17);
    ridList.push_back((RID)10);
    CPPUNIT_ASSERT(m_wrapper.checkIndexValid(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<IdxTupleList>(idxValueList[0]);
    total = testTupleList.size();

    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 102);

    testTuple = testTupleList[3];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 0);

    testTuple = testTupleList[4];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 200);

    rc = m_wrapper.updateIndexRec(m_session.txnid, idxStructList, idxValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    //      CPPUNIT_ASSERT( m_wrapper.deleteIndexRec( idxStructList, idxValueList, ridList ) == NO_ERROR );
  }

  void testInterfaceUpdateMultiColIndex()
  {
    IdxStruct curStruct, testStruct;
    IdxTuple curTuple, testTuple;
    IdxTupleList curTupleList, testTupleList;
    IdxStructList idxStructList;
    IdxValueList idxValueList;
    RIDList ridList;
    std::string curStr, testStr;
    int total, rc;

    testInterfaceCreateIndex();

    CPPUNIT_ASSERT(m_wrapper.updateIndexRec(m_session.txnid, idxStructList, idxValueList, ridList) !=
                   NO_ERROR);

    // test column struct list
    SetIndexStruct(curStruct, 900, 901, 4, WriteEngine::MEDINT);
    idxStructList.push_back(curStruct);

    SetIndexStruct(curStruct, 900, 901, 2, WriteEngine::SMALLINT);
    idxStructList.push_back(curStruct);

    testStruct = idxStructList[0];
    CPPUNIT_ASSERT(testStruct.treeOid == 900);
    CPPUNIT_ASSERT(testStruct.listOid == 901);
    CPPUNIT_ASSERT(testStruct.idxWidth == 4);
    CPPUNIT_ASSERT(testStruct.idxDataType == WriteEngine::MEDINT);

    testStruct = idxStructList[1];
    CPPUNIT_ASSERT(testStruct.treeOid == 900);
    CPPUNIT_ASSERT(testStruct.listOid == 901);
    CPPUNIT_ASSERT(testStruct.idxWidth == 2);
    CPPUNIT_ASSERT(testStruct.idxDataType == WriteEngine::SMALLINT);

    CPPUNIT_ASSERT(m_wrapper.checkIndexValid(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // add values for the first index
    curTuple.data = 102;
    curTupleList.push_back(curTuple);

    curTuple.data = 0;
    curTupleList.push_back(curTuple);

    curTuple.data = 200;
    curTupleList.push_back(curTuple);

    idxValueList.push_back(curTupleList);

    curTupleList.clear();

    // add values for the second index
    curTuple.data = (short)1;
    curTupleList.push_back(curTuple);

    curTuple.data = (short)2;
    curTupleList.push_back(curTuple);

    curTuple.data = (short)3;
    curTupleList.push_back(curTuple);

    idxValueList.push_back(curTupleList);

    curTupleList.clear();

    // add one rowId
    ridList.push_back((RID)5);
    CPPUNIT_ASSERT(m_wrapper.checkIndexValid(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   ERR_ROWID_VALUE_NOT_MATCH);
    ridList.push_back((RID)9);
    ridList.push_back((RID)10);
    CPPUNIT_ASSERT(m_wrapper.checkIndexValid(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<IdxTupleList>(idxValueList[0]);
    total = testTupleList.size();

    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 102);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 0);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 200);

    // second column
    testTupleList = static_cast<IdxTupleList>(idxValueList[1]);
    total = testTupleList.size();

    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<short>(testTuple.data) == 1);

    testTuple = testTupleList[1];
    CPPUNIT_ASSERT(boost::any_cast<short>(testTuple.data) == 2);

    testTuple = testTupleList[2];
    CPPUNIT_ASSERT(boost::any_cast<short>(testTuple.data) == 3);

    rc = m_wrapper.processMultiColIndexRec(m_session.txnid, idxStructList, idxValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
  }

  void testInterfaceDeleteIndexChar()
  {
    IdxStruct curStruct, testStruct;
    IdxTuple curTuple, testTuple;
    IdxTupleList curTupleList, testTupleList;
    std::vector<IdxStruct> idxStructList;
    std::vector<IdxTupleList> idxValueList;
    std::vector<RID> ridList;
    std::string curStr, testStr;
    int total;

    // test column struct list
    SetIndexStruct(curStruct, 900, 901, 4, WriteEngine::MEDINT);
    idxStructList.push_back(curStruct);

    testStruct = idxStructList[0];
    CPPUNIT_ASSERT(testStruct.treeOid == 900);
    CPPUNIT_ASSERT(testStruct.listOid == 901);
    CPPUNIT_ASSERT(testStruct.idxWidth == 4);
    CPPUNIT_ASSERT(testStruct.idxDataType == WriteEngine::MEDINT);

    CPPUNIT_ASSERT(m_wrapper.checkIndexValid(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // add values for the first index
    curTuple.data = 102;
    curTupleList.push_back(curTuple);

    idxValueList.push_back(curTupleList);

    curTupleList.clear();

    // add one rowId
    ridList.push_back((RID)5);
    CPPUNIT_ASSERT(m_wrapper.checkIndexValid(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   NO_ERROR);

    // retrieve the values back
    // first column
    testTupleList = static_cast<IdxTupleList>(idxValueList[0]);
    total = testTupleList.size();

    testTuple = testTupleList[0];
    CPPUNIT_ASSERT(boost::any_cast<int>(testTuple.data) == 102);

    CPPUNIT_ASSERT(m_wrapper.deleteIndexRec(m_session.txnid, idxStructList, idxValueList, ridList) ==
                   NO_ERROR);
  }

  void testInterfaceDctnryToken()
  {
    /*
          int           rc;
          OID           dctnryOID=880, treeOID=881, listOID=882;
          DctnryTuple   dctnryTuple;
          DctnryStruct  dctnryStruct;

          printf("Running testInterfaceDctnryToken\n");
          memset(&dctnryTuple, 0, sizeof(dctnryTuple));
          memset(&dctnryStruct,0, sizeof(dctnryStruct));

          dctnryStruct.dctnryOid = dctnryOID;
          dctnryStruct.treeOid   = treeOID;
          dctnryStruct.listOid   = listOID;

          BRMWrapper::setUseBrm(true);
          m_wrapper.dropStore( m_session.txnid, dctnryOID, treeOID, listOID);
          rc = m_wrapper.createStore( m_session.txnid, dctnryOID, treeOID, listOID);
          CPPUNIT_ASSERT( rc == NO_ERROR );
          rc = m_wrapper.openStore( m_session.txnid, dctnryOID, treeOID, listOID);
          CPPUNIT_ASSERT( rc == NO_ERROR );

           Token token[17], token2[17];
           memset(token, 0,  17*sizeof(Token));
           memset(token2, 0, 17*sizeof(Token));
           int i,j,k,smallSize=1000;
           j=255;
           for (k=1; k<17; k++)
           {
             for (i=0; i<smallSize; i++)
             {
              if (j==0)
               j=255;
              dctnryTuple.sigValue[i]=k;
              j--;
             }
             dctnryTuple.token.fbo=(uint64_t)-1;
             dctnryTuple.token.op =(int)-1;
             dctnryTuple.sigSize = smallSize;
             rc = m_wrapper.tokenize( m_session.txnid, dctnryTuple);
             if (rc!= NO_ERROR)
             {
              printf("1219.Tokenize failed ERROR CODE : %i k %i\n", rc , k);
              return;
             }
             else
             {
                printf("fbo %llu op %i\n",dctnryTuple.token.fbo,dctnryTuple.token.op);
             }
             CPPUNIT_ASSERT( rc == NO_ERROR );
             if ((dctnryTuple.token.fbo ==(uint64_t)-1) ||(dctnryTuple.token.op==(int)-1))
             {
               printf("1440.Tokenize failed ERROR CODE : %i k %i\n", rc , k);
              return;
             }
             token[k].fbo = dctnryTuple.token.fbo;
             token[k].op = dctnryTuple.token.op;

             rc = m_wrapper.tokenize( m_session.txnid, dctnryTuple);
             if (rc!= NO_ERROR)
             {
              printf("k %i \n", k);
              printf("1227.Second time tokenize failed ERROR CODE : %i \n", rc);
              return;
             }
             else
             {
               printf("Second tiem fbo %llu op %i\n",dctnryTuple.token.fbo,dctnryTuple.token.op);
             }
             CPPUNIT_ASSERT( rc == NO_ERROR );
             token2[k].fbo = dctnryTuple.token.fbo;
             token2[k].op  = dctnryTuple.token.op;
           }
           m_wrapper.closeStore( m_session.txnid );
           for (int i=0; i< 17; i++)
           {
             //CPPUNIT_ASSERT(token[i].fbo == token2[i].fbo);
             CPPUNIT_ASSERT(token[i].op  == token2[i].op);
             rc = m_wrapper.deleteToken( m_session.txnid, dctnryStruct, token[i]);
             if (i==0)
              CPPUNIT_ASSERT( rc != NO_ERROR );
             else
              CPPUNIT_ASSERT( rc == NO_ERROR );
           }

          BRMWrapper::setUseBrm( true );

           m_wrapper.openStore( m_session.txnid, dctnryOID, treeOID, listOID);
           rc = m_wrapper.tokenize( m_session.txnid, dctnryStruct,dctnryTuple);
          printf("rc %i \n", rc);
          rc = m_wrapper.rollbackTran(m_session.txnid);
          printf("This is rollback rc %i \n", rc);
          rc = m_wrapper.tokenize( m_session.txnid, dctnryStruct,dctnryTuple);
          printf("This is after rollback tokenize rc %i \n", rc);
          cout << "fbo " << dctnryTuple.token.fbo << " op " << dctnryTuple.token.op << endl;
          m_wrapper.closeStore( m_session.txnid );
         // rc = m_wrapper.dropStore( m_session.txnid, dctnryOID, treeOID, listOID);
         // CPPUNIT_ASSERT( rc == NO_ERROR );
    */
  }
  void testInterfaceDctnryTokenRollBack()
  {
    /*
          int           rc;
          OID           dctnryOID=850, treeOID=851, listOID=852;
          DctnryTuple   dctnryTuple;
          DctnryStruct  dctnryStruct;
          m_session.txnid =100;
          BRMWrapper::setUseBrm(true);
          printf("Running testInterfaceDctnryTokenRollBack\n");
          memset(&dctnryTuple, 0, sizeof(dctnryTuple));
          memset(&dctnryStruct,0, sizeof(dctnryStruct));

          dctnryStruct.dctnryOid = dctnryOID;
          dctnryStruct.treeOid   = treeOID;
          dctnryStruct.listOid   = listOID;

    //      BRMWrapper::setUseBrm(true);
          m_wrapper.dropStore( m_session.txnid, dctnryOID, treeOID, listOID );
          rc = m_wrapper.createStore( m_session.txnid, dctnryOID, treeOID, listOID);
          CPPUNIT_ASSERT( rc == NO_ERROR );
          rc = m_wrapper.openStore( m_session.txnid, dctnryOID, treeOID, listOID);
          CPPUNIT_ASSERT( rc == NO_ERROR );

           Token token[17], token2[17];
           memset(token, 0,  17*sizeof(Token));
           memset(token2, 0, 17*sizeof(Token));

           int i,j,k,smallSize=10;
           j=255;
           for (k=1; k<2; k++)
           {
             for (i=0; i<smallSize; i++)
             {
              if (j==0)
               j=255;
              dctnryTuple.sigValue[i]=k;
              j--;
             }
             dctnryTuple.token.fbo=(uint64_t)-1;
             dctnryTuple.token.op =(int)-1;
             dctnryTuple.sigSize = smallSize;

             rc = m_wrapper.tokenize( m_session.txnid, dctnryTuple);
             if (rc!= NO_ERROR)
             {
              printf("1529.Tokenize failed ERROR CODE : %i k %i\n", rc , k);
              return;
             }
             else
             {
                printf("fbo %llu op %i\n",dctnryTuple.token.fbo,dctnryTuple.token.op);
             }
             CPPUNIT_ASSERT( rc == NO_ERROR );
             if ((dctnryTuple.token.fbo ==(uint64_t)-1) ||(dctnryTuple.token.op==(int)-1))
             {
               printf("1440.Tokenize failed ERROR CODE : %i k %i\n", rc , k);
              return;
             }

           }

          rc = m_wrapper.rollbackTran(m_session.txnid );
          m_wrapper.closeStore( m_session.txnid );
    */
  }

  void testInsertCommitRollback()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    string curString;
    int rc;
    ColumnOp colOp;

    BRMWrapper::setUseBrm(true);
    // test column struct list
    CreateColumnFile(150, 4, WriteEngine::CHAR, WriteEngine::WR_CHAR);

    SetColumnStruct(curColStruct, 150, 4, WriteEngine::CHAR);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 150);
    CPPUNIT_ASSERT(testColStruct.colWidth == 4);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::CHAR);

    // test column values
    // add values for the first column
    curString = "ab1";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "ab2";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    curTupleList.clear();
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    // try to insert more rows and row id should increase
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = m_wrapper.commit(m_session.txnid);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    // txnid increase by one
    m_session.txnid++;

    colValueList.clear();
    curString = "bb1";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "bb2";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = m_wrapper.commit(m_session.txnid);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    // txnid increase by one
    m_session.txnid++;

    colValueList.clear();
    curString = "cb1";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    curString = "cb2";
    curTuple.data = curString;
    curTupleList.push_back(curTuple);

    colValueList.push_back(curTupleList);

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = m_wrapper.rollbackTran(m_session.txnid);
    CPPUNIT_ASSERT(rc == NO_ERROR);
  }

  void testInterfaceInsertRowHwm()
  {
    ColStruct curColStruct, testColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;
    int rc;
    ColumnOp colOp;

    CreateColumnFile(300, 1, WriteEngine::TINYINT, WriteEngine::WR_BYTE);
    CreateColumnFile(400, 4, WriteEngine::INT, WriteEngine::WR_INT);

    // test column struct list
    SetColumnStruct(curColStruct, 300, 1, WriteEngine::TINYINT);
    colStructList.push_back(curColStruct);

    SetColumnStruct(curColStruct, 400, 4, WriteEngine::INT);
    colStructList.push_back(curColStruct);

    testColStruct = colStructList[0];
    CPPUNIT_ASSERT(testColStruct.dataOid == 300);
    CPPUNIT_ASSERT(testColStruct.colWidth == 1);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::TINYINT);

    testColStruct = colStructList[1];
    CPPUNIT_ASSERT(testColStruct.dataOid == 400);
    CPPUNIT_ASSERT(testColStruct.colWidth == 4);
    CPPUNIT_ASSERT(testColStruct.colDataType == WriteEngine::INT);

    m_session.txnid++;
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    // test column values
    // add values for the first column
    for (int i = 0; i < 9000; i++)
    {
      curTuple.data = (char)(i % 20);
      curTupleList.push_back(curTuple);
    }

    colValueList.push_back(curTupleList);

    curTupleList.clear();

    // add values for the second column
    for (int i = 0; i < 9000; i++)
    {
      curTuple.data = i;
      curTupleList.push_back(curTuple);
    }

    colValueList.push_back(curTupleList);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT(ridList.size() == 9000);
  }

  void testInterfaceCombineIndexLoad()
  {
    int rc;
    ColumnOp colOp;
    ColStruct curColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;

    Cache::init();
    Cache::setUseCache(true);

    m_wrapper.setDebugLevel(DEBUG_0);

    colOp.dropColumn(400);
    CreateColumnFile(400, 4, WriteEngine::INT, WriteEngine::WR_INT);

    // test column struct list
    SetColumnStruct(curColStruct, 400, 4, WriteEngine::INT);
    colStructList.push_back(curColStruct);

    m_session.txnid++;
    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) ==
                   ERR_STRUCT_VALUE_NOT_MATCH);

    curTupleList.clear();
    int totalValue = 2000000;

    // add values for the first column
    for (int i = 0; i < totalValue; i++)
    {
      curTuple.data = rand() % 100;
      curTupleList.push_back(curTuple);
    }

    colValueList.push_back(curTupleList);

    CPPUNIT_ASSERT(m_wrapper.checkValid(m_session.txnid, colStructList, colValueList, ridList) == NO_ERROR);

    //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
    rc = m_wrapper.insertColumnRec(m_session.txnid, colStructList, colValueList, ridList);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    CPPUNIT_ASSERT((int)ridList.size() == totalValue);

    if (colOp.exists(900))
    {
      rc = m_wrapper.dropIndex(m_session.txnid, 900, 901);
      CPPUNIT_ASSERT(rc == NO_ERROR);
    }

    rc = m_wrapper.createIndex(m_session.txnid, 900, 901);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    uint64_t totalRows;
    rc = m_wrapper.buildIndex(400, 900, 901, INT, 4, 0, true, totalRows);
    printf("\nrc=%d", rc);
    printf("\ntotalRows=%llu", totalRows);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    Cache::freeMemory();
  }

  void testTmpBulkPerformance()
  {
    int rc;
    ColumnOp colOp;
    ColStruct curColStruct;
    ColTuple curTuple, testTuple;
    ColTupleList curTupleList, testTupleList;
    ColStructList colStructList;
    ColValueList colValueList, colOldValList;
    RIDList ridList;

    Cache::init();
    Cache::setUseCache(true);

    m_wrapper.setDebugLevel(DEBUG_0);

    /*      colOp.dropColumn( 400 );
          CreateColumnFile( 400, 4, WriteEngine::INT, WriteEngine::WR_INT );

          // test column struct list
          SetColumnStruct( curColStruct, 400, 4, WriteEngine::INT );
          colStructList.push_back( curColStruct );

          m_session.txnid++;
          CPPUNIT_ASSERT( m_wrapper.checkValid( m_session.txnid, colStructList, colValueList, ridList ) ==
       ERR_STRUCT_VALUE_NOT_MATCH );

          curTupleList.clear();
          int totalValue = 2000000;
          // add values for the first column
          for( int i = 0; i < totalValue; i++ ) {
             curTuple.data = rand() % 100;
             curTupleList.push_back( curTuple );
          }
          colValueList.push_back( curTupleList );

          CPPUNIT_ASSERT( m_wrapper.checkValid( m_session.txnid, colStructList, colValueList, ridList ) ==
       NO_ERROR );

          //      m_wrapper.printInputValue( (OID)103, colStructList, colValueList, ridList );
          rc = m_wrapper.insertColumnRec( m_session.txnid, colStructList, colValueList, ridList );
          CPPUNIT_ASSERT( rc == NO_ERROR );
          CPPUNIT_ASSERT(  (int)ridList.size() == totalValue );
    */

    if (colOp.exists(900))
    {
      rc = m_wrapper.dropIndex(m_session.txnid, 900, 901);
      CPPUNIT_ASSERT(rc == NO_ERROR);
    }

    rc = m_wrapper.createIndex(m_session.txnid, 900, 901);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    uint64_t totalRows;
    std::string indexName;
    rc = m_wrapper.buildIndex("CPL_299_23839.dat", 900, 901, INT, 4, 0, false, totalRows, indexName, 0,
                              2000000);
    printf("\nrc=%d", rc);
    printf("\ntotalRows=%llu", totalRows);
    CPPUNIT_ASSERT(rc == NO_ERROR);
    Cache::freeMemory();
  }
  void testBoostFloat()
  {
    boost::any anyVal;

    unsigned int fValue = 0xffaaaaaa;
    unsigned int testfValue;
    unsigned int testfValue2;

    float fValue1 = 0xffaaaaaa;
    float testfValue11 = 0;
    float testfValue12 = 0;
    int iSize = 0;
    char charBuf[100];
    FILE* pFile;

    ColumnOp colOp;
    Column curCol;
    int rc;
    RID rowIdArray[3];
    int width = 4, totalRow = 3;
    FID fid = 800;
    float valArray[3];
    uint64_t oldValArray[3];
    CalpontSystemCatalog::ColDataType colDataType = CalpontSystemCatalog::FLOAT;
    ColType colType = WriteEngine::WR_FLOAT;
    // uint64_t          emptyVal = 0xFFAAAAAB;

    rowIdArray[0] = 0;
    rowIdArray[1] = 1;
    rowIdArray[2] = 2;

    valArray[0] = 8.123;
    valArray[1] = 5.345;
    valArray[2] = 3.789;

    CreateColumnFile(fid, width, colDataType, colType);

    colOp.initColumn(curCol);
    colOp.setColParam(curCol, 2, width, colDataType, colType, fid);
    rc = colOp.openColumnFile(curCol);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    rc = colOp.writeRow(curCol, totalRow, (RID*)rowIdArray, valArray, oldValArray);
    CPPUNIT_ASSERT(rc == NO_ERROR);

    for (int i = 0; i < totalRow; i++)
      cout << "emptyVal=" << oldValArray[i] << endl;

    colOp.closeFile(curCol.dataFile.pFile);
    pFile = fopen("/home/jhuang/tmp/000.dir/000.dir/003.dir/FILE032.cdf", "r");
    iSize = fread(charBuf, 1, 4, pFile);
    memcpy(&testfValue12, charBuf, 4);
    cout << "Value1=" << testfValue12 << endl;
    iSize = fread(charBuf, 1, 4, pFile);
    memcpy(&testfValue12, charBuf, 4);
    cout << "Value2=" << testfValue12 << endl;
    iSize = fread(charBuf, 1, 4, pFile);
    memcpy(&testfValue12, charBuf, 4);
    cout << "Value3=" << testfValue12 << endl;
    fclose(pFile);

    anyVal = fValue1;
    testfValue11 = boost::any_cast<float>(anyVal);
    memcpy(charBuf, &testfValue11, 4);
    pFile = fopen("test", "w+b");

    if (pFile != NULL)
    {
      iSize = fwrite((char*)charBuf, 4, 1, pFile);
      printf("\niSize=%d \n", iSize);
      fclose(pFile);
    }

    unsigned char charBuf2[100];
    memset(charBuf2, '\0', 100);
    pFile = fopen("test", "r");
    iSize = fread(charBuf2, 4, 1, pFile);
    memcpy(&testfValue12, charBuf2, 4);

    cout << " Using Float, NULL value=" << testfValue12 << endl;

    anyVal = fValue;
    testfValue = boost::any_cast<unsigned int>(anyVal);
    memcpy(charBuf, &testfValue, 4);
    pFile = fopen("test", "w+b");

    if (pFile != NULL)
    {
      iSize = fwrite((char*)charBuf, 4, 1, pFile);
      printf("\niSize=%d \n", iSize);
      fclose(pFile);
    }

    memset(charBuf2, '\0', 100);
    pFile = fopen("test", "r");
    iSize = fread(charBuf2, 4, 1, pFile);
    memcpy(&testfValue2, charBuf2, 4);
    cout << " Using unsinged int, NULL value=" << testfValue2 << endl;

    // Other method

    anyVal = fValue;
    testfValue = boost::any_cast<unsigned int>(anyVal);
    ofstream fout("file.dat", ios::binary);
    fout.write((char*)(&testfValue), sizeof(testfValue));
    fout.close();

    ifstream fin("file.dat", ios::binary);
    fin.read((char*)(&testfValue2), sizeof(testfValue2));
    cout << "Using unsigned int NULL value=" << testfValue2 << endl;
  }
  void testCreateColumnFiles1()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;
    OID fid = 10;

    m_session.txnid++;
    // file opertaions

    // colOp.startfTimer();
    colOp.initColumn(curCol);
    rc = colOp.createColumn(curCol, 0, 8, WriteEngine::CHAR, WriteEngine::WR_CHAR, (FID)fid);
    // colOp.stopfTimer();
    colOp.clearColumn(curCol);
    // rc = m_wrapper.createColumn( m_session.txnid, fid, WriteEngine::CHAR, 8);

    // cout<< "total run time for 1th column we_char size=8 -> " << colOp.getTotalfRunTime() << " msec"<<
    // endl;
    colOp.deleteFile(fid);
    sleep(3);
  }

  void testCreateColumnFiles2()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;
    OID fid = 20;

    m_session.txnid++;
    // file opertaions

    // colOp.startfTimer();
    colOp.initColumn(curCol);
    rc = colOp.createColumn(curCol, 0, 4, WriteEngine::INT, WriteEngine::WR_CHAR, (FID)fid);
    // colOp.stopfTimer();
    colOp.clearColumn(curCol);
    // rc = m_wrapper.createColumn( m_session.txnid, fid, WriteEngine::INT, 4);
    // colOp.stopfTimer();
    // cout<< "total run time for 2th column we_int size=4 -> " << colOp.getTotalfRunTime() << " msec"<< endl;
    CPPUNIT_ASSERT(rc == NO_ERROR);
    colOp.deleteFile(fid);
    sleep(5);
  }
  void testCreateColumnFiles3()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;
    OID fid = 30;

    m_session.txnid++;
    // file opertaions

    // colOp.startfTimer();
    colOp.initColumn(curCol);
    rc = colOp.createColumn(curCol, 0, 4, WriteEngine::CHAR, WriteEngine::WR_CHAR, (FID)fid);
    colOp.clearColumn(curCol);
    // colOp.stopfTimer();
    // rc = m_wrapper.createColumn( m_session.txnid, fid, WriteEngine::CHAR, 4);
    // colOp.stopfTimer();
    // cout<< "total run time for 3th column we_char size=4 -> " << colOp.getTotalfRunTime() << " msec"<<
    // endl;
    CPPUNIT_ASSERT(rc == NO_ERROR);
    colOp.deleteFile(fid);
    sleep(3);
  }
  void testCreateColumnFiles4()
  {
    ColumnOp colOp;
    Column curCol;
    int rc;
    OID fid = 40;

    m_session.txnid++;
    // file opertaions

    // colOp.startfTimer();
    colOp.initColumn(curCol);
    rc = colOp.createColumn(curCol, 0, 2, WriteEngine::SMALLINT, WriteEngine::WR_CHAR, (FID)fid);
    colOp.clearColumn(curCol);
    // rc = m_wrapper.createColumn( m_session.txnid, fid, WriteEngine::SMALLINT, 2 );
    // colOp.stopfTimer();
    // cout<< "total run time for 4th column we_smallint size=2 -> " << colOp.getTotalfRunTime() << " msec"<<
    // endl;
    CPPUNIT_ASSERT(rc == NO_ERROR);
    colOp.deleteFile(fid);
  }
  void testCreateFileMultipleBF()
  {
    FileOp fileop;
    ColumnOp colOp;
    char fileName[FILE_NAME_SIZE];
    int rc, numOfBlock = 8192, allocSize = 8192;
    FILE* pFile;
    int t_diff;
    struct timeb t_start, t_current;

    OID fid = 600;
    uint64_t emptyVal = 0;
    int width = 4;

    rc = fileop.oid2FileName(fid, fileName, true);
    emptyVal = colOp.getEmptyRowValue(WriteEngine::INT, width);
    int multiple = 1;
    int writeSize = multiple * BYTE_PER_BLOCK;
    unsigned char writeBuf[writeSize];
    BRMWrapper::getInstance()->allocateExtent((const OID)fid, numOfBlock, allocSize);

    pFile = fopen(fileName, "w+b");

    if (pFile != NULL)
    {
      fileop.setEmptyBuf(writeBuf, writeSize, emptyVal, width);
      // setvbuf(pFile , NULL, _IONBF, 0 );
      int loopSize = numOfBlock / multiple;
      ftime(&t_start);

      for (int i = 0; i < loopSize; i++)
      {
        fwrite(writeBuf, writeSize, 1, pFile);
      }

      ftime(&t_current);
      t_diff = (int)(1000.0 * (t_current.time - t_start.time) + (t_current.millitm - t_start.millitm));
      cout << "total run time for create file 64MB with writesize= " << writeSize << " byte"
           << "loop size=" << loopSize << " times"
           << " Time for Multiple Write 64MB with Buffer->we_int size=4 -> " << t_diff << " msec" << endl;

      fileop.closeFile(pFile);
    }

    colOp.deleteFile(fid);
  }
  void testCreateFilelargerBF()
  {
    FileOp fileop;
    ColumnOp colOp;
    char fileName[FILE_NAME_SIZE];
    int rc, numOfBlock = 8192, allocSize = 8192;
    FILE* pFile;

    int t_diff;
    struct timeb t_start, t_current;

    OID fid = 1600;
    uint64_t emptyVal = 0;
    int width = 4;

    rc = fileop.oid2FileName(fid, fileName, true);
    emptyVal = colOp.getEmptyRowValue(WriteEngine::INT, width);
    int multiple = 1;
    int writeSize = multiple * BYTE_PER_BLOCK;
    unsigned char writeBuf[writeSize];
    BRMWrapper::getInstance()->allocateExtent((const OID)fid, numOfBlock, allocSize);

    pFile = fopen(fileName, "w+b");

    if (pFile != NULL)
    {
      fileop.setEmptyBuf(writeBuf, writeSize, emptyVal, width);

      setvbuf(pFile, NULL, _IOFBF, DEFAULT_BUFSIZ);
      // setvbuf(pFile , (char*)writeBuf, _IOFBF, writeSize/*DEFAULT_WRITE_BUFSIZE*/);
      int loopSize = allocSize / multiple;
      ftime(&t_start);

      for (int i = 0; i < loopSize; i++)
      {
        fwrite(writeBuf, writeSize, 1, pFile);
      }

      ftime(&t_current);
      t_diff = (int)(1000.0 * (t_current.time - t_start.time) + (t_current.millitm - t_start.millitm));
      cout << "Buffer size =" << DEFAULT_BUFSIZ << endl;
      cout << "total run time for create file 64MB with writesize= " << writeSize << " byte"
           << "loop size=" << loopSize << " times"
           << " Time for Multiple Write 64MB with Buffer->we_int size=4 -> " << t_diff << " msec" << endl;

      fileop.closeFile(pFile);
    }

    colOp.deleteFile(fid);
  }

  void testCreateFileMultipleIONBF()
  {
    FileOp fileop;
    ColumnOp colOp;
    char fileName[FILE_NAME_SIZE];
    int rc, numOfBlock = 8192, allocSize = 8192;
    FILE* pFile;
    int t_diff;
    struct timeb t_start, t_current;

    OID fid = 400;
    uint64_t emptyVal = 0;
    int width = 4;

    rc = fileop.oid2FileName(fid, fileName, true);
    emptyVal = colOp.getEmptyRowValue(WriteEngine::INT, width);
    int multiple = 128;
    int writeSize = multiple * BYTE_PER_BLOCK;
    unsigned char writeBuf[writeSize];
    BRMWrapper::getInstance()->allocateExtent((const OID)fid, numOfBlock, allocSize);

    pFile = fopen(fileName, "w+b");

    if (pFile != NULL)
    {
      fileop.setEmptyBuf(writeBuf, writeSize, emptyVal, width);
      setvbuf(pFile, NULL, _IONBF, 0);
      int loopSize = numOfBlock / multiple;
      ftime(&t_start);

      for (int i = 0; i < loopSize; i++)
      {
        fwrite(writeBuf, writeSize, 1, pFile);
      }

      ftime(&t_current);
      t_diff = (int)(1000.0 * (t_current.time - t_start.time) + (t_current.millitm - t_start.millitm));
      cout << "total run time for create file 64MB with writesize= " << writeSize << " byte"
           << "loop size=" << loopSize << " times"
           << " Time for Multiple Write 64MB NO Buffer->we_int size=4 -> " << t_diff << " msec" << endl;

      fileop.closeFile(pFile);
    }

    colOp.deleteFile(fid);
  }
  void testCreateFile64MBIONBF()
  {
    FileOp fileop;
    ColumnOp colOp;
    char fileName[FILE_NAME_SIZE];
    int rc, numOfBlock = 8192, allocSize = 8192;
    FILE* pFile;
    int t_diff;
    struct timeb t_start, t_current;

    OID fid = 500;
    uint64_t emptyVal = 0;
    int width = 4;

    rc = fileop.oid2FileName(fid, fileName, true);
    emptyVal = colOp.getEmptyRowValue(WriteEngine::INT, width);
    int multiple = 8192;
    int writeSize = multiple * BYTE_PER_BLOCK;
    unsigned char* writeBuf;
    BRMWrapper::getInstance()->allocateExtent((const OID)fid, numOfBlock, allocSize);

    pFile = fopen(fileName, "w+b");

    if (pFile != NULL)
    {
      writeBuf = new unsigned char[writeSize];
      fileop.setEmptyBuf(writeBuf, writeSize, emptyVal, width);
      setvbuf(pFile, NULL, _IONBF, 0);

      ftime(&t_start);
      fwrite(writeBuf, writeSize, 1, pFile);
      ftime(&t_current);
      t_diff = (int)(1000.0 * (t_current.time - t_start.time) + (t_current.millitm - t_start.millitm));

      cout << "total run time for create file Single Write 64MB NO Buffer->we_int size=4 -> " << t_diff
           << " msec" << endl;
      fileop.closeFile(pFile);
    }

    colOp.deleteFile(fid);
  }
  void testCreateFileNoBRM()
  {
    FileOp fileop;
    ColumnOp colOp;
    char fileName[FILE_NAME_SIZE];
    int rc, allocSize = 8192;

    OID fid = 401;
    uint64_t emptyVal = 0;

    // colOp.deleteFile( fid);

    // cout << "start time=" << colOp.t_start << " msec" << endl;

    rc = fileop.oid2FileName(fid, fileName, true);
    emptyVal = colOp.getEmptyRowValue(WriteEngine::INT, 4);
    // colOp.startfTimer();

    fileop.createFile(fileName, allocSize, emptyVal, 4);
    // colOp.stopfTimer();
    // cout<<"stop time=" << colOp.t_current << " msec"<< endl;
    // cout<< "total run time for column NO BRM we_int size=4 -> " << colOp.getTotalfRunTime() << " msec"<<
    // endl;
    colOp.deleteFile(fid);
  }

  void testSoloman()
  {
    char fileName[80] = "/usr/local/mariadb/columnstore/data1/test.dat";
    char outfile[80] = "/usr/local/mariadb/columnstore/data1/out.dat";
    int numOfBlock = 8192, blockSize = 8192;
    FILE* testFile;
    FILE* outFile;
    int t_diff;
    struct timeb t_start, t_current;
    int multiple = numOfBlock;
    int writeSize = multiple * blockSize;
    unsigned char* writeBuf;
    writeBuf = new unsigned char[writeSize];
    memset(writeBuf, 0, writeSize);
    testFile = fopen(fileName, "w+b");
    outFile = fopen(outfile, "w");

    if (testFile != NULL)
    {
      setvbuf(testFile, (char*)writeBuf, _IOFBF, writeSize);
      int loopSize = numOfBlock / multiple;
      ftime(&t_start);

      for (int i = 0; i < loopSize; i++)
      {
        fwrite(writeBuf, writeSize, 1, testFile);
      }

      ftime(&t_current);
      t_diff = (int)(1000.0 * (t_current.time - t_start.time) + (t_current.millitm - t_start.millitm));
      char buff[256];
      sprintf(buff, " create a 64MB file with writesize= %d byte", writeSize);
      fwrite(buff, strlen(buff), 1, outFile);
      sprintf(buff, " loop size = %d", loopSize);
      fwrite(buff, strlen(buff), 1, outFile);
      sprintf(buff, " Time for writing %d = %d msec", writeSize, t_diff);
      fwrite(buff, strlen(buff), 1, outFile);
      fclose(testFile);
      fclose(outFile);
    }
    else
      fprintf(outFile, "FILE DOES NOT EXIST: %s", fileName);

    delete[] writeBuf;
  }
};

CPPUNIT_TEST_SUITE_REGISTRATION(WriteEngineTest);

#include <cppunit/extensions/TestFactoryRegistry.h>
#include <cppunit/ui/text/TestRunner.h>

int main(int argc, char** argv)
{
  CppUnit::TextUi::TestRunner runner;
  CppUnit::TestFactoryRegistry& registry = CppUnit::TestFactoryRegistry::getRegistry();
  runner.addTest(registry.makeTest());
  bool wasSuccessful = runner.run("", false);
  return (wasSuccessful ? 0 : 1);
}