/* Copyright (C) 2014 InfiniDB, Inc.

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

// $Id: pingproc.cpp 2101 2013-01-21 14:12:52Z rdempsey $
#include <iostream>
#include <iomanip>
#include <sys/types.h>
#include <unistd.h>
#include <getopt.h>
#include <sstream>
#include <vector>
using namespace std;

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

#include "bytestream.h"
using namespace messageqcpp;

#include "distributedenginecomm.h"
#include "primitivemsg.h"
#include "jobstep.h"
#include "batchprimitiveprocessor-jl.h"
using namespace joblist;

#include "calpontsystemcatalog.h"
using namespace execplan;

#include "brm.h"
using namespace BRM;

// Global vars
bool debug;
bool thdFcnFailure;

//
// TODO: Why is this namespace here?

namespace
{
void timespec_sub(const struct timespec& tv1, const struct timespec& tv2, struct timespec& diff)
{
  if (tv2.tv_nsec < tv1.tv_nsec)
  {
    diff.tv_sec = tv2.tv_sec - tv1.tv_sec - 1;
    diff.tv_nsec = tv2.tv_nsec - tv1.tv_nsec + 1000000000;
  }
  else
  {
    diff.tv_sec = tv2.tv_sec - tv1.tv_sec;
    diff.tv_nsec = tv2.tv_nsec - tv1.tv_nsec;
  }
}  // timespec_sub

//
//
class OidOperation
{
 public:
  enum OpType_t
  {
    SCAN = 0,
    BLOCK = 1,
    LOOPBACK = 2,
    NONE = 3,
    BATCHSCAN = 4,
    BATCHSTEP = 5,
    BATCHFILT = 6
  };

  OidOperation(const OID_t oid, const OpType_t opType, const uint32_t sessionId = 0);
  ~OidOperation(){};

  void addFilter(const int8_t COP, const int64_t value);

  const OID_t OID() const
  {
    return fOid;
  }
  const OpType_t OpType() const
  {
    return fOpType;
  }

  const CalpontSystemCatalog::ColType& ColumnType()
  {
    return fColType;
  }
  const uint32_t ColumnWidth() const
  {
    return fColType.colWidth;
  }

  const uint32_t FilterCount() const
  {
    return fFilterCount;
  }
  const ByteStream& FilterString()
  {
    return fFilterList;
  }
  const uint32_t SessionId() const
  {
    return fSessionId;
  }
  const uint32_t DataType() const
  {
    return fColType.colDataType;
  }
  const uint32_t BOP() const
  {
    return fBOP;
  }
  void BOP(const uint32_t bop)
  {
    if (FilterCount() >= 2)
      fBOP = bop;
  }
  const uint32_t COP1() const
  {
    return fCOP1;
  }
  void COP1(const uint32_t cop)
  {
    fCOP1 = cop;
  }
  const uint32_t COP2() const
  {
    return fCOP2;
  }
  void COP2(const uint32_t cop)
  {
    fCOP2 = cop;
  }
  bool isIntegralDataType();
  void setLbidTraceOn();
  void setPMProfileOn();
  bool LbidTrace()
  {
    return fLbidTrace;
  }
  bool PMProfile()
  {
    return fPMProfile;
  }

  void deSerializeFilter(int8_t& COP, int64_t& value);

  // private:
  OidOperation(){};
  OID_t fOid;

  ByteStream fFilterList;
  uint32_t fFilterCount;
  OpType_t fOpType;
  CalpontSystemCatalog::ColType fColType;
  uint32_t fSessionId;
  uint32_t fBOP;
  uint32_t fCOP1;
  uint32_t fCOP2;
  bool fLbidTrace;
  bool fPMProfile;
};  // class OidOperation

//
//
OidOperation::OidOperation(const OID_t oid, const OpType_t opType, const uint32_t sessionId)
 : fOid(oid)
 , fFilterList()
 , fFilterCount(0)
 , fOpType(opType)
 , fSessionId(sessionId)
 , fBOP(BOP_NONE)
 , fCOP1(COMPARE_NIL)
 , fCOP2(COMPARE_NIL)
 , fLbidTrace(false)
 , fPMProfile(false)
{
  boost::shared_ptr<CalpontSystemCatalog> cat =
      execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(getpid());
  fColType = cat->colType(oid);
  fFilterList.reset();
}

void OidOperation::setLbidTraceOn()
{
  fLbidTrace = true;
}

void OidOperation::setPMProfileOn()
{
  fPMProfile = true;
}

void OidOperation::addFilter(const int8_t COP, const int64_t value)
{
  if (fFilterCount == 2)
    return;

  fFilterList << (uint8_t)COP;

  // converts to a type of the appropriate width, then bitwise
  // copies into the filter ByteStream
  switch (ColumnWidth())
  {
    case 1:
      int8_t tmp8;
      tmp8 = value;
      fFilterList << *((uint8_t*)&tmp8);
      break;

    case 2:
      int16_t tmp16;
      tmp16 = value;
      fFilterList << *((uint16_t*)&tmp16);
      break;

    case 4:
      int32_t tmp32;
      tmp32 = value;
      fFilterList << *((uint32_t*)&tmp32);
      break;

    case 8: fFilterList << *((uint64_t*)&value); break;

    default:
      ostringstream o;

      o << "addFilter: colType says OID "
        << " has a width of " << ColumnWidth();
      throw runtime_error(o.str());
  }

  fFilterCount++;
}

void OidOperation::deSerializeFilter(int8_t& COP, int64_t& value)
{
  if (fFilterCount == 0)
  {
    COP = COMPARE_NIL;
    return;
  }

  fFilterList >> *(uint8_t*)&COP;

  switch (ColumnWidth())
  {
    case 1:
      int8_t tmp8;
      tmp8 = value;
      fFilterList >> *((uint8_t*)&tmp8);
      value = tmp8;
      break;

    case 2:
      int16_t tmp16;
      fFilterList >> *((uint16_t*)&tmp16);
      value = tmp16;
      break;

    case 4:
      int32_t tmp32;
      fFilterList >> *((uint32_t*)&tmp32);
      value = tmp32;
      break;

    case 8: fFilterList >> *((uint64_t*)&value); break;

    default:
      ostringstream o;

      o << "deSerializeFilter: colType says OID "
        << " has a width of " << ColumnWidth();
      throw runtime_error(o.str());
  }

  fFilterCount--;
}

typedef vector<OidOperation*> OperationList;

// Only process these column types
//
bool OidOperation::isIntegralDataType()
{
  if (DataType() == CalpontSystemCatalog::BIT || DataType() == CalpontSystemCatalog::TINYINT ||
      DataType() == CalpontSystemCatalog::SMALLINT || DataType() == CalpontSystemCatalog::MEDINT ||
      DataType() == CalpontSystemCatalog::INT || DataType() == CalpontSystemCatalog::DATE ||
      DataType() == CalpontSystemCatalog::BIGINT || DataType() == CalpontSystemCatalog::DATETIME ||
      DataType() == CalpontSystemCatalog::TIMESTAMP || DataType() == CalpontSystemCatalog::TIME ||
      DataType() == CalpontSystemCatalog::UTINYINT || DataType() == CalpontSystemCatalog::USMALLINT ||
      DataType() == CalpontSystemCatalog::UMEDINT || DataType() == CalpontSystemCatalog::UINT ||
      DataType() == CalpontSystemCatalog::UBIGINT)
    return true;

  if (DataType() == CalpontSystemCatalog::CHAR && 1 == fColType.colWidth)
    return true;

  return false;
}

const ByteStream formatLoopBackMsg(const uint32_t sessionId, uint32_t uniqueId)
{
  ByteStream primMsg;
  ISMPacketHeader ism;
  memset(&ism, 0, sizeof(ism));
  ism.Command = COL_LOOPBACK;
  ism.Size = sizeof(ism) + sizeof(ColLoopback);
  ism.Type = 2;
  primMsg.load((const uint8_t*)&ism, sizeof(ism));
  struct ColLoopback lb;

  memset(&lb, 0, sizeof(lb));

  lb.Hdr.SessionID = sessionId;
  lb.Hdr.StatementID = 0;
  lb.Hdr.TransactionID = sessionId;
  lb.Hdr.VerID = 0;
  lb.Hdr.StepID = sessionId;
  lb.Hdr.UniqueID = uniqueId;
  primMsg.append((const uint8_t*)&lb, sizeof(lb));

  return primMsg;
}  // formatLoopBackMsg

const ByteStream formatDictionaryMsg(const uint64_t lbid, ByteStream& ridList, const uint16_t ridCount,
                                     OidOperation& oidOp)
{
  ByteStream primMsg;
  DictSignatureRequestHeader hdr;
  ISMPacketHeader ism;

  ism.Flags = 0;  // planFlagsToPrimFlags(fTraceFlags);
  ism.Command = DICT_SIGNATURE;
  ism.Size = sizeof(DictSignatureRequestHeader) + ridList.length();
  ism.Type = 2;

  hdr.Hdr.SessionID = oidOp.SessionId();
  hdr.Hdr.StatementID = 0;
  hdr.Hdr.TransactionID = oidOp.SessionId();
  hdr.Hdr.VerID = 0;
  hdr.Hdr.StepID = 0;

  hdr.LBID = lbid;
  hdr.PBID = 0;
  idbassert(ridCount <= 8000);
  hdr.NVALS = ridCount;

  primMsg.load((const uint8_t*)&ism, sizeof(ism));
  primMsg.append((const uint8_t*)&hdr, sizeof(DictSignatureRequestHeader));
  primMsg += ridList;

  return primMsg;
}

const ByteStream formatColStepMsg(const uint64_t lbid, ByteStream& ridList, const uint16_t ridCount,
                                  OidOperation& oidOp, uint32_t uniqueId)
{
  ByteStream primMsg;
  NewColRequestHeader hdr;

  memset(&hdr, 0, sizeof(hdr));

  hdr.ism.Reserve = 0;
  hdr.ism.Flags = 0;

  if (oidOp.LbidTrace() == true)
    hdr.ism.Flags |= PF_LBID_TRACE;

  if (oidOp.PMProfile() == true)
    hdr.ism.Flags |= PF_PM_PROF;

  hdr.ism.Command = COL_BY_SCAN;
  hdr.ism.Size = sizeof(NewColRequestHeader) + oidOp.FilterString().length() + ridList.length();
  hdr.ism.Type = 2;

  hdr.hdr.SessionID = oidOp.SessionId();
  hdr.hdr.StatementID = 0;
  hdr.hdr.TransactionID = oidOp.SessionId();
  hdr.hdr.VerID = 0;
  hdr.hdr.StepID = oidOp.SessionId();
  hdr.hdr.UniqueID = uniqueId;

  hdr.LBID = lbid;
  idbassert(hdr.LBID > 0);
  hdr.PBID = 0;
  hdr.DataSize = oidOp.ColumnWidth();
  hdr.DataType = oidOp.DataType();
  hdr.OutputType = OT_BOTH;
  hdr.BOP = BOP_NONE;
  //	hdr.InputFlags = 0;
  hdr.NOPS = oidOp.FilterCount();
  hdr.NVALS = ridCount;
  hdr.sort = 0;

  primMsg.load((const uint8_t*)&hdr, sizeof(NewColRequestHeader));

  if (oidOp.FilterCount() > 0)
    primMsg += oidOp.FilterString();

  if (ridCount > 0)
    primMsg += ridList;

  return primMsg;

}  // formatColStepMsg

const ByteStream formatDictionaryScanMsg(const uint64_t lbid, const uint16_t count, OidOperation& oidOp)
{
  ByteStream primMsg;
  DictTokenByScanRequestHeader hdr;

  hdr.ism.Reserve = 0;
  hdr.ism.Flags = 0;
  hdr.ism.Command = DICT_TOKEN_BY_SCAN_COMPARE;
  hdr.ism.Size = sizeof(DictTokenByScanRequestHeader) + oidOp.FilterString().length();
  hdr.ism.Type = 2;

  hdr.Hdr.SessionID = oidOp.SessionId();
  hdr.Hdr.StatementID = 0;
  hdr.Hdr.TransactionID = 0;
  hdr.Hdr.VerID = 0;
  hdr.Hdr.StepID = oidOp.SessionId();

  hdr.LBID = lbid;
  idbassert(hdr.LBID >= 0);
  hdr.PBID = 0;
  hdr.OutputType = OT_TOKEN;
  hdr.BOP = oidOp.BOP();
  hdr.COP1 = oidOp.COP1();
  hdr.COP2 = oidOp.COP2();
  hdr.NVALS = oidOp.FilterCount();
  hdr.Count = count;
  hdr.charsetNumber = oidOp.ColumnType().charsetNumber;
  idbassert(hdr.Count > 0);

  primMsg.load((const uint8_t*)&hdr.ism, sizeof(ISMPacketHeader));
  primMsg.append((const uint8_t*)&hdr, sizeof(DictTokenByScanRequestHeader));
  primMsg += oidOp.FilterString();

  return primMsg;
}

const ByteStream formatColScanMsg(const uint64_t lbid, const uint16_t count, OidOperation& oidOp,
                                  uint32_t uniqueId)
{
  ByteStream primMsg;
  ISMPacketHeader ism;
  ColByScanRangeRequestHeader fMsgHeader;

  memset(&fMsgHeader, 0, sizeof(fMsgHeader));
  memset(&ism, 0, sizeof(ism));

  ism.Reserve = 0;
  ism.Flags = 0;

  if (oidOp.LbidTrace() == true)
    ism.Flags |= PF_LBID_TRACE;

  if (oidOp.PMProfile() == true)
    ism.Flags |= PF_PM_PROF;

  ism.Command = COL_BY_SCAN_RANGE;
  ism.Size = sizeof(fMsgHeader) + sizeof(ism) + oidOp.FilterString().length();
  ism.Type = 2;

  primMsg.load((const uint8_t*)&ism, sizeof(ism));

  fMsgHeader.LBID = lbid;
  idbassert(fMsgHeader.LBID >= 0);
  fMsgHeader.PBID = 0;
  fMsgHeader.DataSize = oidOp.ColumnWidth();
  fMsgHeader.DataType = oidOp.DataType();
  fMsgHeader.OutputType = OT_BOTH;
  fMsgHeader.BOP = oidOp.BOP();
  fMsgHeader.NOPS = oidOp.FilterCount();
  fMsgHeader.NVALS = 0;
  fMsgHeader.Count = count;  //(hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);
  idbassert(fMsgHeader.Count > 0);
  fMsgHeader.Hdr.SessionID = oidOp.SessionId();
  fMsgHeader.Hdr.StatementID = 0;
  fMsgHeader.Hdr.TransactionID = oidOp.SessionId();
  fMsgHeader.Hdr.VerID = 0;
  fMsgHeader.Hdr.StepID = oidOp.SessionId();
  fMsgHeader.Hdr.UniqueID = uniqueId;

  primMsg.append((const uint8_t*)&fMsgHeader, sizeof(fMsgHeader));

  if (oidOp.FilterCount() > 0)
    primMsg += oidOp.FilterString();

  return primMsg;

}  // formatColScanMsg

//
void doBatchOp_scan(OidOperation& OidOp);
void doBatchOp_step(OidOperation& OidOp);
void doBatchOp_filt(OidOperation& OidOp);
void doBatchQueryOp(OperationList& OidOps);
void doColScan(OidOperation& OidOp);
void doColStep(OidOperation& OidOp);
void doDictScan(OidOperation& OidOp);
void doDictStep(OidOperation& OidOp);

// receive the responses from PrimProc
//
struct ThdFcn
{
  void operator()()
  {
    uint64_t totalBytes = 0;

    try
    {
      ByteStream ibs;

      if (debug)
        cout << "Waiting on " << fNumMsgs << " messages." << endl;

      for (uint32_t k = 0; k < fNumMsgs; k++)
      {
        // 				cout << "reading msg #" << k << "...\n";
        ibs = fDec->read(uniqueID);

        // 				cout << "got msg #" << k << endl;
        if (debug)
          if (k % 10240 == 0)
            cout << "ThdFcn: read " << fSessionid << " " << k << "/" << fNumMsgs << " " << ibs.length() << "/"
                 << totalBytes << endl;

        if (ibs.length() == 0)
          break;

        totalBytes += ibs.length();
      }
    }
    catch (exception& e)
    {
      cerr << "read exception: " << e.what() << endl;
      thdFcnFailure = true;
    }

    if (debug)
      cout << totalBytes << " bytes read in " << fNumMsgs << " messages" << endl;

  }  // void operator()

  uint32_t fSessionid;
  uint32_t uniqueID;
  DistributedEngineComm* fDec;
  unsigned fNumMsgs;

};  // struct ThdFcn

struct QryThdFcn
{
  void operator()()
  {
    uint64_t totalBytes = 0;
    int64_t min;
    int64_t max;
    uint64_t lbid;
    uint32_t cachedIO;
    uint32_t physIO;
    uint32_t touchedBlocks;
    bool validCPData;

    try
    {
      ByteStream ibs;
      ByteStream obs;

      if (debug)
        cout << "Waiting on " << fNumMsgs << " messages." << endl;

      for (uint32_t k = 0; k < fNumMsgs; k++)
      {
        // 				cout << "reading msg #" << k << "...\n";
        ibs = fDec->read(uniqueID);

        // 				cout << "got msg #" << k << endl;
        if (debug)
          if (k % 10240 == 0)
            cout << "QryThdFcn: read " << fSessionid << " " << k << "/" << fNumMsgs << " " << ibs.length()
                 << "/" << totalBytes << " rows: " << fRows << endl;

        if (ibs.length() == 0)
          break;

        totalBytes += ibs.length();
        fRows +=
            fBpp.getTableBand(ibs, &obs, &validCPData, &lbid, &min, &max, &cachedIO, &physIO, &touchedBlocks);
        fBlockTouched += touchedBlocks;
      }
    }
    catch (exception& e)
    {
      cerr << "read exception: " << e.what() << endl;
      thdFcnFailure = true;
    }

    if (debug)
      cout << totalBytes << " bytes read in " << fNumMsgs << " messages for " << fRows << " rows and "
           << fBlockTouched << " blocks\n";

  }  // void operator()

  QryThdFcn(BatchPrimitiveProcessorJL& bpp, uint64_t& rows, uint32_t& blk)
   : fBpp(bpp), fNumMsgs(0), fRows(rows), fBlockTouched(blk)
  {
  }
  BatchPrimitiveProcessorJL& fBpp;
  uint32_t fSessionid;
  DistributedEngineComm* fDec;
  unsigned fNumMsgs;
  uint64_t& fRows;
  uint32_t& fBlockTouched;
  uint32_t uniqueID;

};  // struct ThdFcn

struct BatchScanThr
{
  BatchScanThr(OidOperation& oidOp) : fOidOp(oidOp)
  {
  }

  void operator()()
  {
    doBatchOp_scan(fOidOp);
  }  // void operator()

  OidOperation& fOidOp;

};  // struct BatchScanThr

struct BatchStepThr
{
  BatchStepThr(OidOperation& oidOp) : fOidOp(oidOp)
  {
  }

  void operator()()
  {
    doBatchOp_step(fOidOp);
  }  // void operator()

  OidOperation& fOidOp;

};  // struct BatchStepThr

struct BatchFiltThr
{
  BatchFiltThr(OidOperation& oidOp) : fOidOp(oidOp)
  {
  }

  void operator()()
  {
    doBatchOp_filt(fOidOp);
  }  // void operator()

  OidOperation& fOidOp;

};  // struct BatchFiltThr

struct BatchQueryThr
{
  BatchQueryThr(OperationList& oidOps) : fOidOps(oidOps)
  {
  }

  void operator()()
  {
    doBatchQueryOp(fOidOps);
  }  // void operator()

  OperationList& fOidOps;

};  // struct

struct ColStepThr
{
  ColStepThr(OidOperation& oidOp) : fOidOp(oidOp)
  {
  }

  void operator()()
  {
    doColStep(fOidOp);
  }  // void operator()

  OidOperation& fOidOp;

};  // struct ColStepThr

struct ColScanThr
{
  ColScanThr(OidOperation& oidOp) : fOidOp(oidOp)
  {
  }

  void operator()()
  {
    doColScan(fOidOp);
  }  // void operator()

  OidOperation& fOidOp;

};  // struct ColStepThr

struct DictSigThr
{
};  // DictSigThr

struct DictScanThr
{
};  // DictScanThr

// doColScan
void doDictionaryScan(OidOperation& OidOp)
{
}

// doColScan
void doColScan(OidOperation& OidOp)
{
  if (debug)
    cout << "beginning doColScan\n";

  BRM::LBIDRange_v lbidRanges;
  HWM_t hwm = 0;
  ResourceManager* rm = ResourceManager::instance();
  DistributedEngineComm* dec = DistributedEngineComm::instance(rm);
  struct timespec ts1;
  struct timespec ts2;
  struct timespec diff;
  uint32_t uniqueID;

  uint32_t sessionid = getpid();
  uint32_t totalBlks = 0;
  // 	dec->addSession(sessionid);
  // 	dec->addStep(sessionid, sessionid);
  DBRM dbrm;

  uniqueID = dbrm.getUnique32();
  dec->addQueue(uniqueID);

  int err = dbrm.lookup(OidOp.OID(), lbidRanges);

  if (err)
    throw runtime_error("doAColScan: BRM LBID range lookup failure (1)");

  err = dbrm.getHWM(OidOp.OID(), hwm);

  if (err)
    throw runtime_error("doAColScan: BRM HWM lookup failure (3)");

  LBIDRange_v::iterator it;
  OID_t tmp;
  uint32_t fbo;

  ThdFcn f1;
  f1.fSessionid = sessionid;
  f1.uniqueID = uniqueID;
  f1.fDec = dec;
  f1.fNumMsgs = 0;
  thdFcnFailure = false;
  uint32_t rangeSize = 0;
  ByteStream obs;

  // calculate the expected number of messages
  for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
  {
    BRM::LBID_t lbid = (*it).start;

    if (dbrm.lookup(lbid, 0, false, tmp, fbo))
    {
      cerr << "pColScanStep::sendPrimitiveMessages: dbrm.lookup failed for lbid " << lbid << endl;
      abort();
    }

    if (hwm < fbo)
      continue;

    rangeSize = (hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);
    totalBlks += rangeSize;
  }  // for

  f1.fNumMsgs = (totalBlks / OidOp.ColumnWidth());

  if (0 < totalBlks % OidOp.ColumnWidth())
    ++f1.fNumMsgs;

  idbassert(f1.fNumMsgs);

  if (debug)
    cout << "Scanning OID " << OidOp.OID() << " " << f1.fNumMsgs << " msgs" << endl;

  thread t1(f1);
  clock_gettime(CLOCK_REALTIME, &ts1);
  // send the primitive requests
  int rCount = 0;
  totalBlks = 0;

  for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
  {
    try
    {
      BRM::LBID_t lbid = (*it).start;

      if (dbrm.lookup(lbid, 0, false, tmp, fbo))
      {
        cerr << "doAColScan dbrm.lookup failed for lbid " << lbid << endl;
        abort();
      }

      if (hwm < fbo)
        continue;

      rangeSize = (hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);
      obs = formatColScanMsg(lbid, rangeSize, OidOp, uniqueID);

      if (obs.length() > 0)
      {
        dec->write(obs);
        rCount++;

        if (debug)
          cout << "colScan: " << rCount << "/" << lbidRanges.size() << " sending " << obs.length()
               << " bytes "
               << " lbid " << lbid << " sz " << rangeSize << endl;
      }
    }
    catch (exception& e)
    {
      cerr << "catch " << e.what() << endl;
    }

    totalBlks += rangeSize;

  }  // for (lbidRanges ...

  t1.join();
  clock_gettime(CLOCK_REALTIME, &ts2);
  // 	dec->removeSession(sessionid);
  dec->removeQueue(uniqueID);

  timespec_sub(ts1, ts2, diff);

  cout << "ColScan stats OID: " << OidOp.OID() << "\tFilter: " << (int)OidOp.FilterCount()
       << "\tBlocks: " << (int)totalBlks << "\tElapse: " << diff.tv_sec + (diff.tv_nsec / 1000000000.0)
       << "s";

  float rate = 0;
  rate = totalBlks / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << " Blks/sec : " << rate << endl;

  if (thdFcnFailure)
    cout << "There was a failure in the read thread." << endl;

  cout << endl;

}  // doAScan

//
void usage()
{
  cerr << "PingProc operation [filter] [operation] [filter] [reporting]" << endl
       << "PingProc -c -s <oid> -gt 0 -t <oid> -d " << endl
       << "\t---- operation flags ----" << endl
       << "\t--scan -s <scan the oid>" << endl
       << "\t--block -t <colstep the oid>" << endl
       << "\t--BatchPrimitiveScan -B <batch scan the oid>" << endl
       << "\t--BatchPrimitiveStep -Z <batch step the oid>" << endl
       << "\t--concurrent -c perform each operation in its own thread" << endl
       << "\t--lbid-trace -lb set lbid trace flag in request" << endl
       << "\t---- filter flags ----" << endl
       << "\t--equal -eq <int> equivalency test of values in a block>" << endl
       << "\t--greater-than -gt  <int> - greater than (>) test of values in a block" << endl
       << "\t--greater-than-equal -ge <int> greater than or equal to (>=) test of values in a block" << endl
       << "\t--less-than -lt <int> less than (<) test of values in a block" << endl
       << "\t--less-than-equal -le <int> less than or equal to (<=) test of values in a block" << endl
       << "\t--not-equal -ne <int> not equal to (!=) test of values in a block" << endl
       << "\t--bop <1 or 0> binary operator when 2 comparison filters are present" << endl
       << "\t---- reporting flags ----" << endl
       << "\t--debug -d turn debug output on>" << endl
       << "\t--list <oid> -list <oid> -l <oid> print out all oids and their ranges" << endl
       << "\t--loopback <count=100000> -p <count> send count loopback requests" << endl
       << "\t--query -q run batch query for all the oids (enter with -B or -Z)" << endl;

}  // usage()

const int64_t getInt(string s)
{
  if (s.length() <= 0)
    return -1;

  // if (atoll(s.data()) < 0)
  //	return -1;

  return atoll(s.data());

}  // getInt

// dictionary
void doDictionarySig(OidOperation& OidOp)
{
}  // doDictionarySig

// col step
void doColStep(OidOperation& OidOp)
{
  struct timespec ts1;
  struct timespec ts2;
  struct timespec diff;
  DBRM dbrm;
  BRM::LBIDRange_v lbidRanges;
  HWM_t hwm = 0;
  LBIDRange_v::iterator it;
  OID_t tmp;
  uint32_t fbo;
  uint32_t totalBlks = 0;
  ResourceManager* rm = ResourceManager::instance();
  DistributedEngineComm* dec = DistributedEngineComm::instance(rm);
  ThdFcn f1;

  // 	dec->addSession(OidOp.SessionId());
  // 	dec->addStep(OidOp.SessionId(), OidOp.SessionId());
  uint32_t uniqueID = dbrm.getUnique32();
  dec->addQueue(uniqueID);

  f1.fSessionid = OidOp.SessionId();
  f1.uniqueID = uniqueID;
  f1.fDec = dec;
  f1.fNumMsgs = 0;
  thdFcnFailure = false;
  ByteStream ridlist;
  uint16_t ridCount = 0;  // BLOCK_SIZE/OidOp.ColumnWidth();

  for (uint16_t i = 0; i < ridCount; i++)
    ridlist << i;

  int err = dbrm.lookup(OidOp.OID(), lbidRanges);

  if (err)
    throw runtime_error("doAColStep: BRM LBID range lookup failure (1)");

  err = dbrm.getHWM(OidOp.OID(), hwm);

  if (err)
    throw runtime_error("doAColStep: BRM HWM lookup failure (3)");

  uint32_t rangeSize = 0;

  for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
  {
    BRM::LBID_t lbid = (*it).start;

    if (dbrm.lookup(lbid, 0, false, tmp, fbo))
    {
      cerr << "pColStep::sendPrimitiveMessages: dbrm.lookup failed for lbid " << lbid << endl;
      abort();
    }

    if (hwm < fbo)
      break;  // continue;

    rangeSize = (hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);
    totalBlks += rangeSize;
  }  // for

  uint32_t colwidth = OidOp.ColumnWidth();
  f1.fNumMsgs = totalBlks / (colwidth);

  if (0 < totalBlks % colwidth)
    ++f1.fNumMsgs;

  idbassert(f1.fNumMsgs);
  thread t1(f1);
  ByteStream obs;
  totalBlks = 0;
  clock_gettime(CLOCK_REALTIME, &ts1);

  for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
  {
    BRM::LBID_t lbid = (*it).start;

    if (dbrm.lookup(lbid, 0, false, tmp, fbo))
    {
      if (debug)
        cerr << "pColScanStep::sendPrimitiveMessages: dbrm.lookup failed for lbid " << lbid << endl;

      abort();
    }

    if (hwm < fbo)
      break;  // continue;

    rangeSize = (hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);

    for (unsigned i = 0; i < rangeSize; i++)
    {
      obs += formatColStepMsg(lbid + i, ridlist, ridCount, OidOp, uniqueID);

      if (0 == (i + 1) % colwidth)
      {
        dec->write(obs);

        if (debug && i + 1 == rangeSize)
          cout << "colStep: " << i << "/" << rangeSize << " " << obs.length() << " lbid " << lbid + i << endl;

        obs.restart();
      }
    }

    totalBlks += rangeSize;
  }  // for

  if (obs.length())
  {
    dec->write(obs);

    if (debug)
      cout << "colStep: last"
           << "/" << rangeSize << " " << obs.length() << endl;
  }

  obs.reset();

  t1.join();  //@bug 849 moved join here and changed output to be like pColScan.
  clock_gettime(CLOCK_REALTIME, &ts2);
  timespec_sub(ts1, ts2, diff);
  // 	t1.join();
  cout << "ColStep stats OID: " << OidOp.OID() << "\tFilter: " << (int)OidOp.FilterCount()
       << "\tBlocks: " << (int)totalBlks << "\tElapse: " << diff.tv_sec + (diff.tv_nsec / 1000000000.0)
       << "s";

  float rate = 0;
  rate = totalBlks / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << "\tBlks/sec " << rate << endl;

  if (thdFcnFailure)
    cerr << "There was a failure in the read thread." << endl;

  cout << endl;

}  // doColStep

void doBatchOp_scan(OidOperation& OidOp)
{
  struct timespec ts1, ts2, diff;
  JobStepAssociation injs, outjs;
  ResourceManager* rm = ResourceManager::instance();
  DistributedEngineComm* dec = DistributedEngineComm::instance(rm);
  ThdFcn f1;
  boost::shared_ptr<CalpontSystemCatalog> sysCat =
      execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(getpid());

  pColScanStep scan(injs, outjs, dec, sysCat, OidOp.fOid, OidOp.fOid, OidOp.fSessionId, 0, OidOp.fSessionId,
                    OidOp.fSessionId, OidOp.fSessionId, rm);

  int32_t filters = OidOp.FilterCount();

  while (OidOp.FilterCount() > 0)
  {
    int8_t cop;
    int64_t value;
    OidOp.deSerializeFilter(cop, value);
    scan.addFilter(cop, value);
  }

  BatchPrimitiveProcessorJL bpp;
  ByteStream bs;
  DBRM dbrm;
  BRM::LBIDRange_v lbidRanges;
  HWM_t hwm = 0;
  LBIDRange_v::iterator it;
  OID_t tmp;
  uint32_t fbo;

  uint32_t uniqueID = dbrm.getUnique32();
  bpp.setUniqueID(uniqueID);

  bpp.setSessionID(OidOp.SessionId());
  bpp.setStepID(OidOp.SessionId());
  bpp.addFilterStep(scan);

  cout << "session number = " << OidOp.SessionId() << endl;
  // 	dec->addSession(OidOp.SessionId());
  // 	dec->addStep(OidOp.SessionId(), OidOp.SessionId());
  dec->addQueue(uniqueID);
  f1.fSessionid = OidOp.SessionId();
  f1.uniqueID = uniqueID;
  f1.fDec = dec;
  thdFcnFailure = false;
  int err = dbrm.lookup(OidOp.OID(), lbidRanges);

  if (err)
  {
    cerr << "doAColScan: BRM LBID range lookup failure (1)\n";
    throw runtime_error("doAColScan: BRM LBID range lookup failure (1)");
  }

  err = dbrm.getHWM(OidOp.OID(), hwm);

  if (err)
  {
    cerr << "doAColScan: BRM HWM lookup failure (3)" << endl;
    throw runtime_error("doAColScan: BRM HWM lookup failure (3)");
  }

  f1.fNumMsgs = hwm / OidOp.fColType.colWidth + (hwm % OidOp.fColType.colWidth ? 1 : 0);

  thread t1(f1);

  bpp.createBPP(bs);
  dec->write(bs);
  bs.restart();

  uint32_t rangeSize = 0, totalBlks = 0;
  clock_gettime(CLOCK_REALTIME, &ts1);

  // 	cout << "BPP scaning\n";
  for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
  {
    BRM::LBID_t lbid = (*it).start;

    if (dbrm.lookup(lbid, 0, false, tmp, fbo))
    {
      cerr << "pColScanStep::sendPrimitiveMessages: dbrm.lookup failed for lbid " << lbid << endl;
      abort();
    }

    if (hwm < fbo)
      continue;

    rangeSize = (hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);
    bpp.setLBID(lbid);
    bpp.setCount(rangeSize / OidOp.fColType.colWidth + (rangeSize % OidOp.fColType.colWidth ? 1 : 0));
    bpp.runBPP(bs);
    dec->write(bs);
    // 		cout << "sending the BPP\n";
    bpp.reset();
    bs.restart();
    totalBlks += rangeSize;
  }

  t1.join();
  clock_gettime(CLOCK_REALTIME, &ts2);
  timespec_sub(ts1, ts2, diff);
  float rate = 0;
  cout << "ColStep stats OID: " << OidOp.OID() << " " << (diff.tv_sec + (diff.tv_nsec / 1000000000.0)) << "s"
       << "\tFilters: " << filters << "\tBlocks : " << (int)totalBlks;

  rate = totalBlks / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << "\tBlks/sec " << rate << endl;

  if (thdFcnFailure)
    cerr << "There was a failure in the read thread." << endl;

  bpp.destroyBPP(bs);
  dec->write(bs);
  cout << endl;
}

void doBatchOp_filt(OidOperation& OidOp)
{
  struct timespec ts1, ts2, diff;
  JobStepAssociation injs, outjs;
  ResourceManager* rm = ResourceManager::instance();
  DistributedEngineComm* dec = DistributedEngineComm::instance(rm);
  ThdFcn f1;
  boost::shared_ptr<CalpontSystemCatalog> sysCat = CalpontSystemCatalog::makeCalpontSystemCatalog(getpid());

  ByteStream bs;
  DBRM dbrm;
  BRM::LBIDRange_v lbidRanges;
  HWM_t hwm = 0;
  LBIDRange_v::iterator it;
  OID_t tmp;
  uint32_t fbo;
  uint32_t uniqueID;

  BatchPrimitiveProcessorJL bpp;
  uniqueID = dbrm.getUnique32();
  bpp.setUniqueID(uniqueID);
  bpp.setSessionID(OidOp.SessionId());
  bpp.setStepID(OidOp.SessionId());

  pColScanStep scan(injs, outjs, dec, sysCat, OidOp.fOid, OidOp.fOid, OidOp.fSessionId, 0, OidOp.fSessionId,
                    OidOp.fSessionId, OidOp.fSessionId, rm);

  while (OidOp.FilterCount() > 0)
  {
    int8_t cop;
    int64_t value;
    OidOp.deSerializeFilter(cop, value);
    scan.addFilter(cop, value);
  }

  bpp.addFilterStep(scan);

  pColStep step(injs, outjs, dec, sysCat, OidOp.fOid + 1, OidOp.fOid + 1, OidOp.fSessionId, 0,
                OidOp.fSessionId, OidOp.fSessionId, OidOp.fSessionId, rm);

  while (OidOp.FilterCount() > 0)
  {
    int8_t cop;
    int64_t value;
    OidOp.deSerializeFilter(cop, value);
    step.addFilter(cop, value);
  }

  bpp.addFilterStep(step);

  execplan::CalpontSystemCatalog::ColType colType;
  FilterStep filt(OidOp.fSessionId, OidOp.fSessionId, OidOp.fSessionId, colType);
  filt.setBOP(OidOp.BOP());
  bpp.addFilterStep(filt);

  cout << "session number = " << OidOp.SessionId() << endl;
  // 	dec->addSession(OidOp.SessionId());
  // 	dec->addStep(OidOp.SessionId(), OidOp.SessionId());
  dec->addQueue(uniqueID);
  f1.fSessionid = OidOp.SessionId();
  f1.uniqueID = uniqueID;
  f1.fDec = dec;
  thdFcnFailure = false;
  int err = dbrm.lookup(OidOp.OID(), lbidRanges);

  if (err)
  {
    cerr << "doBatchOp_filt: BRM LBID range lookup failure (1)\n";
    throw runtime_error("doAColScan: BRM LBID range lookup failure (1)");
  }

  err = dbrm.getHWM(OidOp.OID(), hwm);

  if (err)
  {
    cerr << "doBatchOp_filt: BRM HWM lookup failure (2)" << endl;
    throw runtime_error("doBatchOp_filt: BRM HWM lookup failure (2)");
  }

  f1.fNumMsgs = hwm / OidOp.fColType.colWidth + (hwm % OidOp.fColType.colWidth ? 1 : 0);

  thread t1(f1);

  bpp.createBPP(bs);
  dec->write(bs);
  bs.restart();

  uint32_t rangeSize = 0, totalBlks = 0;
  clock_gettime(CLOCK_REALTIME, &ts1);

  // 	cout << "BPP scaning\n";
  for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
  {
    BRM::LBID_t lbid = (*it).start;

    if (dbrm.lookup(lbid, 0, false, tmp, fbo))
    {
      cerr << "doBatchOp_filt: dbrm.lookup failed for lbid (3)" << lbid << endl;
      abort();
    }

    if (hwm < fbo)
      continue;

    rangeSize = (hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);
    bpp.setLBID(lbid);
    bpp.setCount(rangeSize / OidOp.fColType.colWidth + (rangeSize % OidOp.fColType.colWidth ? 1 : 0));
    bpp.runBPP(bs);
    dec->write(bs);
    // 		cout << "sending the BPP\n";
    bpp.reset();
    bs.restart();
    totalBlks += rangeSize;
  }

  t1.join();
  clock_gettime(CLOCK_REALTIME, &ts2);
  timespec_sub(ts1, ts2, diff);
  float rate = 0;
  cout << "doBatchOp_filt stats OID: " << OidOp.OID() << " " << (diff.tv_sec + (diff.tv_nsec / 1000000000.0))
       << "s"
       << "\tBlocks : " << (int)totalBlks;

  rate = totalBlks / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << "\tBlks/sec " << rate << endl;

  if (thdFcnFailure)
    cerr << "There was a failure in the read thread." << endl;

  bpp.destroyBPP(bs);
  dec->write(bs);
  cout << endl;
}

void doBatchQueryOp(OperationList& OidOps)
{
  struct timespec ts1, ts2, diff;

  JobStepAssociation injs, outjs;
  BatchPrimitiveProcessorJL bpp;
  uint64_t rows = 0;
  uint32_t blockTouched = 0;
  DBRM dbrm;

  QryThdFcn f1(bpp, rows, blockTouched);

  OperationList::iterator filterOp = OidOps.begin();
  uint32_t sessionId = (*filterOp)->SessionId();
  uint32_t uniqueID = dbrm.getUnique32();
  bpp.setUniqueID(uniqueID);
  bpp.setSessionID(sessionId);
  bpp.setStepID(sessionId);
  cout << "session number = " << sessionId << endl;

  f1.fSessionid = sessionId;
  ResourceManager* rm = ResourceManager::instance();
  DistributedEngineComm* dec = DistributedEngineComm::instance(rm);
  // 	dec->addSession(sessionId);
  // 	dec->addStep(sessionId, sessionId);
  dec->addQueue(uniqueID);

  f1.fDec = dec;
  f1.uniqueID = uniqueID;
  //	boost::shared_ptr<CalpontSystemCatalog> sysCat =
  //execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(getpid());

  // first column is made into the first scan filter step including filters
  OID_t scanOid = (*filterOp)->fOid;
  uint32_t scanWidth = (*filterOp)->ColumnWidth();
  uint32_t maxWidth = scanWidth;

  uint32_t pid = getpid();
  pColScanStep scan(injs, outjs, dec, execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(pid), scanOid,
                    scanOid, sessionId, 0, sessionId, sessionId, sessionId, rm);

  uint32_t filterCount = (*filterOp)->FilterCount();

  while ((*filterOp)->FilterCount() > 0)
  {
    int8_t cop;
    int64_t value;
    (*filterOp)->deSerializeFilter(cop, value);
    scan.addFilter(cop, value);
  }

  bpp.addFilterStep(scan);

  // Any other columns that are batch scans are added as filter steps, the rest as project steps.
  // The last filter step is added as a passthru step into the project list.

  OperationList::iterator listend = OidOps.end();

  for (OperationList::iterator op = OidOps.begin() + 1; op != listend; ++op)
  {
    pColStep step(injs, outjs, dec, execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(pid),
                  (*op)->fOid, (*op)->fOid, sessionId, 0, sessionId, sessionId, sessionId, rm);

    if ((*op)->OpType() == OidOperation::BATCHSCAN)
    {
      filterCount += (*op)->FilterCount();

      while ((*op)->FilterCount() > 0)
      {
        int8_t cop;
        int64_t value;
        (*op)->deSerializeFilter(cop, value);
        step.addFilter(cop, value);
      }

      filterOp = op;
      bpp.addFilterStep(step);
    }
    else
    {
      bpp.addProjectStep(step);
    }

    if ((*op)->ColumnWidth() > maxWidth)
      maxWidth = (*op)->ColumnWidth();
  }

  PassThruStep pass(injs, outjs, dec, (*filterOp)->ColumnType(), (*filterOp)->fOid, (*filterOp)->fOid,
                    sessionId, 0, sessionId, sessionId, sessionId, false, rm);
  bpp.addProjectStep(pass);

  ByteStream bs;
  BRM::LBIDRange_v lbidRanges;
  HWM_t hwm = 0;
  LBIDRange_v::iterator it;
  OID_t tmp;
  uint32_t fbo;

  thdFcnFailure = false;
  int err = dbrm.lookup(scanOid, lbidRanges);

  if (err)
  {
    cerr << "doQueryScan: BRM LBID range lookup failure (1)\n";
    throw runtime_error("doQueryScan: BRM LBID range lookup failure (1)");
  }

  err = dbrm.getHWM(scanOid, hwm);

  if (err)
  {
    cerr << "doQueryScan: BRM HWM lookup failure (3)" << endl;
    throw runtime_error("doQueryScan: BRM HWM lookup failure (3)");
  }

  f1.fNumMsgs = hwm / scanWidth + (hwm % scanWidth ? 1 : 0);

  thread t1(f1);
  uint32_t cnt = dbrm.getExtentSize() / maxWidth;

  bpp.createBPP(bs);
  dec->write(bs);
  bs.restart();
  uint32_t rangeSize = 0, totalBlks = 0;
  clock_gettime(CLOCK_REALTIME, &ts1);

  // 	cout << "BPP scaning\n";
  for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
  {
    BRM::LBID_t lbid = (*it).start;

    if (dbrm.lookup(lbid, 0, false, tmp, fbo))
    {
      cerr << "doBatchQuery dbrm.lookup failed for lbid " << lbid << endl;
      abort();
    }

    if (hwm < fbo)
      continue;

    rangeSize = (hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);
    uint32_t totallbid = rangeSize / scanWidth + (0 < rangeSize % scanWidth ? 1 : 0);

    while (0 < totallbid)
    {
      if (totallbid < cnt)
        cnt = totallbid;

      bpp.setLBID(lbid);
      bpp.setCount(cnt);
      bpp.runBPP(bs);
      dec->write(bs);
      //		    cout << "sending the BPP with range cnt " << cnt  << " lbid " << lbid << "\n";
      bpp.reset();
      bs.restart();
      lbid += cnt * scanWidth;
      totallbid -= cnt;
    }

    for (OperationList::iterator op = OidOps.begin(); op != OidOps.end(); ++op)
    {
      totalBlks += (uint32_t)(rangeSize * (double)((double)(*op)->ColumnWidth() / scanWidth));
    }
  }

  t1.join();
  clock_gettime(CLOCK_REALTIME, &ts2);
  timespec_sub(ts1, ts2, diff);

  float rate = 0;
  cout << "QueryScan stats - " << bpp.toString()
       << "\tElapsed: " << (diff.tv_sec + (diff.tv_nsec / 1000000000.0)) << "s"
       << "\tFilters: " << filterCount << "\tBlocks : " << (int)totalBlks;

  rate = totalBlks / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << "\tBlks/sec " << rate << endl;
  cout << "\tTouched Blocks: " << blockTouched;
  rate = blockTouched / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << "\tTouched Blks/sec " << rate << endl;

  cout << "\tRows: " << rows;
  rate = rows / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << "\t\tRows/sec " << fixed << setprecision(2) << rate << endl;

  if (thdFcnFailure)
    cerr << "There was a failure in the read thread." << endl;

  bpp.destroyBPP(bs);
  dec->write(bs);
  cout << endl;
}

void doBatchOp_step(OidOperation& OidOp)
{
  struct timespec ts1, ts2, diff;
  DBRM dbrm;
  BRM::LBIDRange_v lbidRanges;
  HWM_t hwm = 0;
  LBIDRange_v::iterator it;
  OID_t tmp;
  uint32_t fbo;
  uint32_t totalBlks = 0;
  ResourceManager* rm = ResourceManager::instance();
  DistributedEngineComm* dec = DistributedEngineComm::instance(rm);
  ThdFcn f1;
  JobStepAssociation injs, outjs;

  boost::shared_ptr<CalpontSystemCatalog> sysCat =
      execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(getpid());

  pColStep step(injs, outjs, dec, sysCat, OidOp.fOid, OidOp.fOid, OidOp.fSessionId, 0, OidOp.fSessionId,
                OidOp.fSessionId, OidOp.fSessionId, rm);

  int32_t filters = OidOp.FilterCount();

  while (OidOp.FilterCount() > 0)
  {
    int8_t cop;
    int64_t value;
    OidOp.deSerializeFilter(cop, value);
    step.addFilter(cop, value);
  }

  BatchPrimitiveProcessorJL bpp;
  ElementType et;
  ByteStream obs;
  uint32_t uniqueID = dbrm.getUnique32();
  bpp.setUniqueID(uniqueID);
  bpp.setSessionID(OidOp.SessionId());
  bpp.setStepID(OidOp.SessionId());
  bpp.addFilterStep(step);

  // 	dec->addSession(OidOp.SessionId());
  // 	dec->addStep(OidOp.SessionId(), OidOp.SessionId());
  dec->addQueue(uniqueID);
  f1.fSessionid = OidOp.SessionId();
  f1.uniqueID = uniqueID;
  f1.fDec = dec;
  f1.fNumMsgs = 0;
  thdFcnFailure = false;
  ByteStream ridlist;
  uint16_t ridCount = 0;  // BLOCK_SIZE/OidOp.ColumnWidth();

  for (uint16_t i = 0; i < ridCount; i++)
    ridlist << i;

  int err = dbrm.lookup(OidOp.OID(), lbidRanges);

  if (err)
    throw runtime_error("doAColScan: BRM LBID range lookup failure (1)");

  err = dbrm.getHWM(OidOp.OID(), hwm);

  if (err)
    throw runtime_error("doAColScan: BRM HWM lookup failure (3)");

  uint32_t rangeSize = 0;
  f1.fNumMsgs = hwm / OidOp.fColType.colWidth + (hwm % OidOp.fColType.colWidth ? 1 : 0);
  thread t1(f1);

  totalBlks = 0;
  bpp.createBPP(obs);
  dec->write(obs);
  obs.restart();

  clock_gettime(CLOCK_REALTIME, &ts1);

  for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
  {
    BRM::LBID_t lbid = (*it).start;

    if (dbrm.lookup(lbid, 0, false, tmp, fbo))
    {
      if (debug)
        cerr << "pColScanStep::sendPrimitiveMessages: dbrm.lookup failed for lbid " << lbid << endl;

      abort();
    }

    if (hwm < fbo)
      continue;

    rangeSize = (hwm > (fbo + (*it).size - 1) ? (*it).size : hwm - fbo + 1);

    for (unsigned i = 0; i < rangeSize; i++)
    {
      /* insert all rids for this LBID */
      for (uint32_t j = 0; j < BLOCK_SIZE / OidOp.fColType.colWidth; ++j)
      {
        et.first = ((fbo + i) * BLOCK_SIZE / OidOp.fColType.colWidth) + j;
        et.second = j;
        bpp.addElementType(et);
      }

      /* If on a logical block boundary, send the primitive */
      if (i % OidOp.fColType.colWidth == (unsigned)OidOp.fColType.colWidth - 1)
      {
        // 				cout << "serializing at extent offset " << i << endl;
        bpp.runBPP(obs);
        dec->write(obs);
        bpp.reset();
        obs.restart();
      }

      if (debug && i + 1 == rangeSize)
        cout << "colStep: " << i + 1 << "/" << rangeSize << " " << obs.length() << " lbid " << lbid + i
             << endl;
    }

    if (rangeSize % OidOp.fColType.colWidth)
    {
      // 			cout << "serializing last msg\n";
      bpp.runBPP(obs);
      dec->write(obs);
      bpp.reset();
      obs.restart();
    }

    totalBlks += rangeSize;
  }  // for

  t1.join();
  clock_gettime(CLOCK_REALTIME, &ts2);
  timespec_sub(ts1, ts2, diff);
  float rate = 0;
  cout << "ColStep stats OID: " << OidOp.OID() << " " << (diff.tv_sec + (diff.tv_nsec / 1000000000.0)) << "s"
       << "\tFilters: " << filters << "\tBlocks : " << (int)totalBlks;

  rate = totalBlks / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << "\tBlks/sec " << rate << endl;

  if (thdFcnFailure)
    cerr << "There was a failure in the read thread." << endl;

  bpp.destroyBPP(obs);
  dec->write(obs);

  cout << endl;
}

//
//
void doListOp(const OID_t o = 0)
{
  DBRM dbrm;
  BRM::LBIDRange_v lbidRanges;
  LBIDRange_v::iterator it;
  OID_t oid = 3000;
  HWM_t hwm = 0;

  if (o != 0)
  {
    int err = dbrm.lookup(o, lbidRanges);

    if (err)
      throw runtime_error("doAColScan: BRM LBID range lookup failure (1)");

    err = dbrm.getHWM(o, hwm);

    if (err)
      throw runtime_error("doAColScan: BRM HWM lookup failure (3)");

    cout << "Object ID: " << o << " HWM: " << hwm << endl;

    for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
      cout << "\tStart: " << (*it).start << " sz: " << (*it).size << endl;
  }
  else
  {
    for (; oid < 100000; oid++)
    {
      int err = dbrm.lookup(oid, lbidRanges);

      if (lbidRanges.size() == 0)
        continue;

      if (err)
        throw runtime_error("doAColScan: BRM LBID range lookup failure (1)");

      err = dbrm.getHWM(oid, hwm);

      if (err)
        throw runtime_error("doAColScan: BRM HWM lookup failure (3)");

      cout << "Object ID: " << oid << " HWM: " << hwm << endl;

      for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
        cout << "\tStart: " << (*it).start << " sz: " << (*it).size << endl;

      hwm = 0;
      lbidRanges.clear();
    }  // for (; oid...
  }    // else

}  // doListOp

//
// do LoopBackOp

void doLoopBack(const uint64_t loopcount)
{
  ByteStream lbMsg;
  struct timespec ts1;
  struct timespec ts2;
  struct timespec diff;
  uint32_t sessionid = getpid();
  DBRM dbrm;
  ResourceManager* rm = ResourceManager::instance();
  DistributedEngineComm* dec = DistributedEngineComm::instance(rm);
  ThdFcn f1;

  // 	dec->addSession(sessionid);
  // 	dec->addStep(sessionid, sessionid);
  uint32_t uniqueID = dbrm.getUnique32();
  dec->addQueue(uniqueID);
  f1.fSessionid = sessionid;
  f1.uniqueID = uniqueID;
  f1.fDec = dec;
  f1.fNumMsgs = loopcount;
  thdFcnFailure = false;
  thread t1(f1);
  lbMsg = formatLoopBackMsg(sessionid, uniqueID);

  cout << "Sending " << loopcount << " LOOPBACK requests" << endl;
  clock_gettime(CLOCK_REALTIME, &ts1);

  for (uint64_t i = 0; i < loopcount; i++)
  {
    lbMsg = formatLoopBackMsg(sessionid, uniqueID);
    dec->write(lbMsg);
  }

  clock_gettime(CLOCK_REALTIME, &ts2);
  cout << loopcount << " LOOPBACK msgs sent" << endl;

  t1.join();
  timespec_sub(ts1, ts2, diff);
  cout << "\ttotal runtime: " << (diff.tv_sec + (diff.tv_nsec / 1000000000.0)) << "s" << endl;

  float rate = 0;
  rate = loopcount / (diff.tv_sec + (diff.tv_nsec / 1000000000.0));
  cout << "\t" << rate << " rqsts/s" << endl;
  dec->removeQueue(uniqueID);

}  // doLoopBack

}  // namespace

/*---------------------------------------------------------------------------
//Command line parameter definition
//
// -o <oid1> -o <oid2> -o . . . <oidN>
// -s <scan the oid>
// -t <colstep the oid>
// -d <turn debug output on>
// -eq -equal <int> < == test of values in a block>
// -gt -greater-than <int> < greater than (>) test of values in a block>
// -ge -greater-than-equal <int> < greater than or equal to (>=) test of values in a block>
// -lt -less-than <int> < less than (<) test of values in a block>
// -le -less-than-equal <int> < less than or equal to (<=) test of values in a block>
// -ne -not-equal <int> < not equal to (!=) test of values in a block>
// -bop <1 or 0> <1 AND 0 OR binary operator when 2 comparison filters are present>
// --list -list -l <optional oid> <print out all oids or the specified oid and their ranges>
// --loopback -loopback <count> send <count> loopback requests
// --concurrent run all jobs currently if set to true. defaults to false
// --lbid-trace -lb turn on lbid tracing
// --pm-profile -pmp turn on pm profiling
-----------------------------------------------------------------------------*/

int main(int argc, char** argv)
{
  int64_t eq_val = 0;
  int64_t gt_val = 0;
  int64_t ge_val = 0;
  int64_t lt_val = 0;
  int64_t le_val = 0;
  int64_t ne_val = 0;
  int64_t bop_val = 0;
  int64_t loopback_count = 100000;
  int64_t scanOid = 0;
  int64_t stepOid = 0;
  bool list = false;
  // OID_t listOid=0;
  string oidString;
  vector<OID_t> oidv;
  int ch = 0;
  bool concurrent_flag = false;
  bool lbidtrace_flag = false;
  bool pmprofile_flag = false;
  bool query_flag = false;

  enum CLA_ENUM
  {
    OID = (int)0,
    SCANOP = (int)1,
    BLOCKOP = (int)2,
    DEBUG = (int)3,
    EQFILTER = (int)4,
    GTFILTER = (int)5,
    GEFILTER = (int)6,
    LTFILTER = (int)7,
    LEFILTER = (int)8,
    NEFILTER = (int)9,
    BOP = (int)10,
    LISTOP = (int)11,
    LOOPBACKOP = (int)12,
    CONCURRENT = (int)13,
    LBIDTRACE = (int)14,
    PMPROFILE = (int)15,
    INVALIDOP = (int)16,
    BATCHSCANOP = (int)17,
    BATCHSTEPOP = (int)18,
    BATCHFILTOP = (int)19,
    QUERYOP = (int)20

  };

  /**
  // longopt struct
  struct option {
      const char *name;
      int has_arg;
      int *flag;
      int val;
  };
  **/

  static struct option long_options[] = {
      //	{const char *name, 		int has_arg, 		int *flag,	int val},
      {"scan", required_argument, NULL, SCANOP},
      {"block", required_argument, NULL, BLOCKOP},
      {"debug", no_argument, NULL, DEBUG},
      {"equal", required_argument, NULL, EQFILTER},
      {"eq", required_argument, NULL, EQFILTER},
      {"greater-than", required_argument, NULL, GTFILTER},
      {"gt", required_argument, NULL, GTFILTER},
      {"greater-than-equal", required_argument, NULL, GEFILTER},
      {"ge", required_argument, NULL, GEFILTER},
      {"less-than", required_argument, NULL, LTFILTER},
      {"lt", required_argument, NULL, LTFILTER},
      {"less-than-equal", required_argument, NULL, LEFILTER},
      {"le", required_argument, NULL, LEFILTER},
      {"not-equal", required_argument, NULL, NEFILTER},
      {"ne", required_argument, NULL, NEFILTER},
      {"bop", optional_argument, NULL, BOP},
      {"list", no_argument, NULL, LISTOP},
      {"loopback", optional_argument, NULL, LOOPBACKOP},
      {"concurrent", no_argument, NULL, CONCURRENT},
      {"lbid-trace", no_argument, NULL, LBIDTRACE},
      {"lb", no_argument, NULL, LBIDTRACE},
      {"pm-prof", no_argument, NULL, PMPROFILE},
      {"pm-profile", no_argument, NULL, PMPROFILE},
      {"pmp", no_argument, NULL, PMPROFILE},
      {"batchscan", required_argument, NULL, BATCHSCANOP},
      {"batchstep", required_argument, NULL, BATCHSTEPOP},
      {"batchfilt", required_argument, NULL, BATCHFILTOP},
      {"queryop", no_argument, NULL, QUERYOP},
      {0, 0, 0, 0}};

  OidOperation* currOp = NULL;
  OperationList OpList;

  if (argc <= 1)
  {
    usage();
  }

  // process command line arguments
  while ((ch = getopt_long_only(argc, argv, "B:Z:F:ds:t:lcqp:", long_options, NULL)) != -1)
  {
    pid_t pidId = getpid();

    switch (ch)
    {
      case SCANOP:
      case 's':
        if (optarg)
          scanOid = getInt(optarg);

        // cout << "OPT=" << ch << " ARG " << scanOid << endl;
        currOp = NULL;

        if (scanOid > 0)
          currOp = new OidOperation(scanOid, OidOperation::SCAN, pidId);
        else
        {
          cout << "PingProc: scan missing or invalid OID parameter value" << endl;
          break;
        }

        if (currOp && currOp->isIntegralDataType())
          OpList.push_back(currOp);
        else
        {
          cout << "PingProc cannot process this ColumnType-oid: " << scanOid << endl;
          delete currOp;
          currOp = NULL;
        }

        break;

      case BATCHSCANOP:
      case 'B':
        if (optarg)
          scanOid = getInt(optarg);
        else
          cout << "no optarg\n";

        cout << "OPT=" << ch << " ARG " << scanOid << endl;
        currOp = NULL;

        if (scanOid > 0)
          currOp = new OidOperation(scanOid, OidOperation::BATCHSCAN, pidId);
        else
        {
          cout << "PingProc: batch scan missing or invalid OID parameter value" << endl;
          break;
        }

        if (currOp && currOp->isIntegralDataType())
          OpList.push_back(currOp);
        else
        {
          cout << "PingProc cannot process this ColumnType-oid: " << scanOid << endl;
          delete currOp;
          currOp = NULL;
        }

        break;

      case BATCHSTEPOP:
      case 'Z':
        if (optarg)
          scanOid = getInt(optarg);
        else
          cout << "no optarg\n";

        cout << "OPT=" << ch << " ARG " << scanOid << endl;
        currOp = NULL;

        if (scanOid > 0)
          currOp = new OidOperation(scanOid, OidOperation::BATCHSTEP, pidId);
        else
        {
          cout << "PingProc: batch step missing or invalid OID parameter value" << endl;
          break;
        }

        if (currOp && currOp->isIntegralDataType())
          OpList.push_back(currOp);
        else
        {
          cout << "PingProc cannot process this ColumnType-oid: " << scanOid << endl;
          delete currOp;
          currOp = NULL;
        }

        break;

      case BATCHFILTOP:
      case 'F':
        if (optarg)
          scanOid = getInt(optarg);
        else
          cout << "no optarg\n";

        cout << "OPT=" << ch << " ARG " << scanOid << endl;
        currOp = NULL;

        if (scanOid > 0)
          currOp = new OidOperation(scanOid, OidOperation::BATCHFILT, pidId);
        else
        {
          cout << "PingProc: batch filter missing or invalid OID parameter value" << endl;
          break;
        }

        if (currOp && currOp->isIntegralDataType())
          OpList.push_back(currOp);
        else
        {
          cout << "PingProc cannot process this ColumnType-oid: " << scanOid << endl;
          delete currOp;
          currOp = NULL;
        }

        break;

      case BLOCKOP:
      case 't':
        if (optarg)
          stepOid = getInt(optarg);

        // cout << "OPT=" << ch << " ARG " << stepOid << endl;
        currOp = NULL;

        if (stepOid > 0)
          currOp = new OidOperation(stepOid, OidOperation::BLOCK, pidId);
        else
        {
          cout << "PingProc: step missing or invalid OID parameter value" << endl;
          break;
        }

        if (currOp && currOp->isIntegralDataType())
          OpList.push_back(currOp);
        else
        {
          cout << "PingProc cannot process this ColumnType-oid: " << stepOid << endl;
          delete currOp;
          currOp = NULL;
        }

        break;

      case DEBUG:
      case 'd':
        // cout << "OPT=" << ch << endl;
        debug = true;
        break;

      case EQFILTER:
        if (optarg)
          eq_val = getInt(optarg);
        else
          eq_val = 0;

        cout << "OPT=" << ch << " ARG=" << eq_val << endl;

        if (currOp)
          currOp->addFilter(COMPARE_EQ, eq_val);
        else
          ;  // TODO: error Processing

        break;

      case GTFILTER:
        if (optarg)
          gt_val = getInt(optarg);
        else
          gt_val = 0;

        // cout << "OPT=" << ch << " ARG=" << gt_val << endl;
        if (currOp)
          currOp->addFilter(COMPARE_GT, gt_val);

        break;

      case GEFILTER:
        if (optarg)
          ge_val = getInt(optarg);
        else
          ge_val = 0;

        // cout << "OPT=" << ch << " ARG=" << ge_val << endl;
        if (currOp)
          currOp->addFilter(COMPARE_GE, ge_val);

        break;

      case LTFILTER:
        if (optarg)
          lt_val = getInt(optarg);
        else
          lt_val = 0;

        // cout << "OPT=" << ch << " ARG=" << lt_val << endl;
        if (currOp)
          currOp->addFilter(COMPARE_LT, lt_val);

        break;

      case LEFILTER:
        if (optarg)
          le_val = getInt(optarg);
        else
          le_val = 0;

        // cout << "OPT=" << ch << " ARG=" << le_val << endl;
        if (currOp)
          currOp->addFilter(COMPARE_LE, le_val);

        break;

      case NEFILTER:
      case 'n':
        if (optarg)
          ne_val = getInt(optarg);
        else
          ne_val = 0;

        // cout << "OPT=" << ch << " ARG=" << ne_val << endl;
        if (currOp)
          currOp->addFilter(COMPARE_NE, ne_val);

        break;

      case BOP:
      case 'b':
        if (optarg)
          bop_val = getInt(optarg);
        else
          bop_val = 1;  // assume AND

        // cout << "OPT=" << ch << " ARG=" << bop_val << endl;
        if (currOp)
          currOp->BOP(bop_val ? BOP_AND : BOP_OR);

        break;

      case LISTOP:
      case 'l':
        /**
        if (optarg)
                listOid=getInt(optarg);
        else
                listOid=0;
        cout << "OPT=" << ch << " LISTOP " << listOid << endl;
        **/
        list = true;
        break;

      case LOOPBACKOP:
      case 'p':
        if (optarg)
          loopback_count = getInt(optarg);

        // cout << "OPT=" << ch << " LOOPBACKOP " << loopback_count << endl;
        currOp = NULL;
        currOp = new OidOperation(0, OidOperation::LOOPBACK, pidId);
        OpList.push_back(currOp);
        break;

      case 'c':
      case CONCURRENT:
        concurrent_flag = true;
        // cout << "OPT=" << ch << " CONCURRENT " << concurrent_flag << endl;
        break;

      case LBIDTRACE:
        lbidtrace_flag = true;

        // cout << "OPT=" << ch << " CONCURRENT " << concurrent_flag << endl;
        if (currOp)
          currOp->setLbidTraceOn();

        break;

      case PMPROFILE:
        pmprofile_flag = true;

        // cout << "OPT=" << ch << " CONCURRENT " << concurrent_flag << endl;
        if (currOp)
          currOp->setPMProfileOn();

        break;

      case 'q':
      case QUERYOP:
        query_flag = true;
        // 			cout << "OPT=" << ch << " QUERY FLAG " << query_flag << endl;
        break;

      case '?':
      default: cout << "optarg " << optarg << endl; usage();
    }

    if (list == true)
      break;

  }  // while

  // if list is requested, print the listing and exit
  //
  vector<struct BatchScanThr*> BatchScanThreads;
  vector<struct BatchStepThr*> BatchStepThreads;
  vector<struct BatchFiltThr*> BatchFiltThreads;
  vector<struct ColScanThr*> ColScanThreads;
  vector<struct ColStepThr*> ColStepThreads;
  vector<struct DictScanThr*> DictScanThreads;
  vector<struct DictSigThr*> DictSigThreads;
  vector<thread*> thrArray;

  if (query_flag)
  {
    cout << "starting batch query thread\n";
    struct BatchQueryThr* qt = new struct BatchQueryThr(OpList);
    thread* t1 = new thread(*qt);

    if (concurrent_flag)
      thrArray.push_back(t1);
    else
      t1->join();
  }
  else if (list)
  {
    doListOp();
  }
  else
  {
    for (uint32_t i = 0; i < OpList.size(); i++)
    {
      if (OpList[i]->OpType() == OidOperation::LOOPBACK)
      {
        doLoopBack(loopback_count);
      }

      else if (OpList[i]->OpType() == OidOperation::SCAN)
      {
        struct ColScanThr* cst = new struct ColScanThr(*OpList[i]);
        ColScanThreads.push_back(cst);
        thread* t1 = new thread(*cst);

        if (concurrent_flag)
          thrArray.push_back(t1);
        else
        {
          t1->join();
          delete t1;
        }
      }

      else if (OpList[i]->OpType() == OidOperation::BLOCK)
      {
        struct ColStepThr* cst = new struct ColStepThr(*OpList[i]);
        ColStepThreads.push_back(cst);
        thread* t1 = new thread(*cst);

        if (concurrent_flag)
          thrArray.push_back(t1);
        else
        {
          t1->join();
          delete t1;
        }
      }

      else if (OpList[i]->OpType() == OidOperation::BATCHSCAN)
      {
        cout << "starting batch scan thread\n";
        struct BatchScanThr* cst = new struct BatchScanThr(*OpList[i]);
        BatchScanThreads.push_back(cst);
        thread* t1 = new thread(*cst);

        if (concurrent_flag)
          thrArray.push_back(t1);
        else
        {
          t1->join();
          delete t1;
        }
      }

      else if (OpList[i]->OpType() == OidOperation::BATCHSTEP)
      {
        cout << "starting batch step thread\n";
        struct BatchStepThr* cst = new struct BatchStepThr(*OpList[i]);
        BatchStepThreads.push_back(cst);
        thread* t1 = new thread(*cst);

        if (concurrent_flag)
          thrArray.push_back(t1);
        else
        {
          t1->join();
          delete t1;
        }
      }

      else if (OpList[i]->OpType() == OidOperation::BATCHFILT)
      {
        cout << "starting batch filt thread\n";
        struct BatchFiltThr* cst = new struct BatchFiltThr(*OpList[i]);
        BatchFiltThreads.push_back(cst);
        thread* t1 = new thread(*cst);

        if (concurrent_flag)
          thrArray.push_back(t1);
        else
        {
          t1->join();
          delete t1;
        }

        i += 2;
      }
    }  // for

  }  // else

  // join threads to main
  for (uint32_t i = 0; i < thrArray.size(); i++)
    thrArray[i]->join();

  // clean up
  for (uint32_t i = 0; i < OpList.size(); i++)
    delete OpList[i];

  OpList.clear();

  for (uint32_t i = 0; i < BatchScanThreads.size(); i++)
    delete BatchScanThreads[i];

  BatchScanThreads.clear();

  for (uint32_t i = 0; i < BatchStepThreads.size(); i++)
    delete BatchStepThreads[i];

  BatchStepThreads.clear();

  for (uint32_t i = 0; i < BatchFiltThreads.size(); i++)
    delete BatchFiltThreads[i];

  BatchFiltThreads.clear();

  for (uint32_t i = 0; i < ColScanThreads.size(); i++)
    delete ColScanThreads[i];

  ColScanThreads.clear();

  for (uint32_t i = 0; i < ColStepThreads.size(); i++)
    delete ColStepThreads[i];

  ColStepThreads.clear();

  for (uint32_t i = 0; i < DictScanThreads.size(); i++)
    delete DictScanThreads[i];

  DictScanThreads.clear();

  for (uint32_t i = 0; i < DictSigThreads.size(); i++)
    delete DictSigThreads[i];

  DictSigThreads.clear();

  for (uint32_t i = 0; i < thrArray.size(); i++)
    delete thrArray[i];

  thrArray.clear();

}  // main()