text: /* 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 <unistd.h>
  #include <cstdlib>
  #include <cstdio>
  #include <cstring>
  #include <ctime>
  
  #include <vector>
  #include <string>
  #include <sstream>
  #include <iostream>
  #include <exception>
  #include <stdexcept>
  #include <cerrno>
  using namespace std;
  
  #include <boost/uuid/uuid.hpp>
  #include <boost/uuid/uuid_generators.hpp>
  #include <boost/uuid/uuid_io.hpp>
  #include <boost/filesystem.hpp>
  
  #include "dataconvert.h"
  #include "liboamcpp.h"
  using namespace oam;
  
  #include "we_cmdargs.h"
  
  #include "installdir.h"
  #include "mcsconfig.h"
  
  namespace WriteEngine
  {
  
  //----------------------------------------------------------------------
  //----------------------------------------------------------------------
  WECmdArgs::WECmdArgs(int argc, char** argv) :
      fMultiTableCount(0),
      fJobLogOnly(false),
      fHelp(false),
      fMode(1),
      fArgMode(-1),
      fQuiteMode(true),
      fConsoleLog(false),
      fVerbose(0),
      fBatchQty(10000),
      fNoOfReadThrds(0),
      fDebugLvl(0),
      fMaxErrors(-1),
      fReadBufSize(0),
      fIOReadBufSize(0),
      fSetBufSize(0),
      fColDelim('|'),
      fEnclosedChar(0),
      fEscChar(0),
      fNoOfWriteThrds(0),
      fNullStrMode(false),
      fImportDataMode(IMPORT_DATA_TEXT),
      fCpiInvoke(false),
      fBlockMode3(false),
      fbTruncationAsError(false),
      fUUID(boost::uuids::nil_generator()()),
      fConsoleOutput(true),
      fTimeZone("SYSTEM"),
      fErrorDir(string(MCSLOGDIR)+"/cpimport/")
  {
      try
      {
          appTestFunction();
          parseCmdLineArgs(argc, argv);
      }
      catch (std::exception& exp)
      {
          std::string exceptMsg( exp.what() );
          exceptMsg += "\nTry 'cpimport -h' for more information.";
          throw (runtime_error( exceptMsg ));
      }
  }
  
  //----------------------------------------------------------------------
  
  void WECmdArgs::appTestFunction()
  {
  
      // testing begins
      //std::string aJobFile("/home/bpaul/Calpont/data/bulk/job/Job_1234.xml");
      //std::string aJobFile("/usr/local/mariadb/columnstore/data/bulk/job/Job_1234.xml");
      //setSchemaAndTableFromJobFile(aJobFile);
      //setEnclByAndEscCharFromJobFile(aJobFile);
      //exit(1);
  
      //testing ends
      return;
  }
  
  //----------------------------------------------------------------------
  std::string WECmdArgs::getCpImportCmdLine()
  {
      std::ostringstream aSS;
      std::string aCmdLine;
  
  
      aSS << fPrgmName;	//prgm name as arg[0]
  
      if (fHelp)
      {
          aSS << " -h ";
          aCmdLine = aSS.str();
          return aCmdLine;
      }
  
      //checkJobIdCase();			// check if JobID
  
  
      if ((fPmFile.length() > 0) && (0 == getMode()))
          aSS << " -l " << fPmFile;
  
      //BUG 4379 if -m is not given while prep args, default is m=1 but
      //on single node -m will set it to 3, after prep args.
      //if((fPmFilePath.length()>0)&&(1!=getMode()))
      //	aSS << " -f " << fPmFilePath;
      if ((fPmFilePath.length() > 0) && (1 != getMode()))
      {
          if (fPmFilePath == "STDIN")	//if stdin, must pass it
              aSS << " -f " << fPmFilePath;
          else if (2 == getArgMode())	//Mode 2 we need to pass the -f option
              aSS << " -f " << fPmFilePath;
          else if (3 == getArgMode()) 	//-m given, -f built in already.
              aSS << " -f " << fPmFilePath;
          else if (0 == fLocFile.length()) //No filename given, from job file
              aSS << " -f " << fPmFilePath;
      }
  
      if (fErrorDir.length() > 0)
          aSS << " -L " << fErrorDir;
  
      if (fUsername.length() > 0)
          aSS << " -U " << fUsername;
  
      if (fJobId.length() > 0)
          aSS << " -j " << fJobId;
  
      if (fNoOfReadThrds > 0)
          aSS << " -r " << fNoOfReadThrds;
  
      if (fNoOfWriteThrds > 0)
          aSS << " -w " << fNoOfWriteThrds;
  
      if (fMaxErrors >= 0)
          aSS << " -e " << fMaxErrors;
  
      // BUG 5088
      if (fDebugLvl > 0)
          aSS << " -d " << fDebugLvl;
  
      if (fSetBufSize > 0)
          aSS << " -B " << fSetBufSize;
  
      if (fColDelim != '|')
      {
          if (fColDelim == '\t')
              aSS << " -s " << "\\t"; //- working with user '\t'
          // NONE of the following will work.
          //aSS << " -s " << "\t"; //aSS << " -s " << "\"\\t\""; //aSS << " -s " << "'\\t'";
          else
              aSS << " -s " << fColDelim;
      }
  
      if (fEnclosedChar != 0)
          aSS << " -E " << fEnclosedChar;
  
      if (fEscChar != 0)
          aSS << " -C " << fEscChar;
  
      if (fNullStrMode)
          aSS << " -n " << '1';
  
      if (fImportDataMode != IMPORT_DATA_TEXT)
          aSS << " -I " << fImportDataMode;
  
      //if(fConfig.length()>0)
      //	aSS << " -c " << fConfig;
      if (fReadBufSize > 0)
      {
          cout << "setting fReadBufSize = " << fReadBufSize;
          aSS << " -c " << fReadBufSize;
      }
  
      if (fIOReadBufSize > 0)
          aSS << " -b " << fIOReadBufSize;
  
  
      if ((fJobPath.length() > 0) && (fMode == 3))
          aSS << " -p " << fJobPath;
  
  
      if (fConsoleLog)
          aSS << " -i ";
  
      if ((fMode == 1) || (fMode == 2))
      {
          aSS << " -R " << getBrmRptFileName();
          aSS << " -m " << fMode;
      }
  
      aSS << " -P " << getModuleID();
      aSS << " -T " << fTimeZone;
  
      if (fbTruncationAsError)
          aSS << " -S ";
  
      if (!fS3Key.empty() && !(fMode == 0 || fMode == 1))
      {
          if (fS3Secret.empty() || fS3Bucket.empty() || fS3Region.empty())
              throw (runtime_error("Not all required S3 options provided"));
          aSS << " -y " << fS3Key;
          aSS << " -K " << fS3Secret;
          aSS << " -t " << fS3Bucket;
          aSS << " -g " << fS3Region;
  
          if (!fS3Host.empty())
          {
              aSS << " -H " << fS3Host;
          }
      }
  
      if ((fJobId.length() > 0) && (fMode == 1) && (!fJobLogOnly))
      {
          // if JobPath provided, make it w.r.t WES
          aSS << " -p " << fTmpFileDir;
          aSS << " -fSTDIN";
      }
      else if ((fJobId.length() > 0) && (fMode == 2) && (!fJobLogOnly))
      {
          // if JobPath provided, make it w.r.t WES
          aSS << " -p " << fTmpFileDir;
  
          if (fPmFile.length() > 0)
              aSS << " -l " << fPmFile;
  
          if (fPmFilePath.length() > 0)
              aSS << " -f " << fPmFilePath;
      }
      else	// do not provide schema & table with JobId
      {
  
          if (!fUUID.is_nil())
              aSS << " -u" << boost::uuids::to_string(fUUID);
  
          if (fSchema.length() > 0)
              aSS << " " << fSchema;
          //else if((fMode != 0)||(fMode==3))	//TODO make it mode3 + jobID
          else if (fJobId.length() > 0)
          { }// may or may not contain Schema.
          //else if((fMode == 1)||(fMode==2))	//TODO make it mode3 + jobID
          else if (fMode != 0)
              throw (runtime_error("Schema not available"));
  
          if (fTable.length() > 0)
              aSS << " " << fTable;
          else if (fJobId.length() > 0)
          {} //may or may not contain Table.
          else if (fMode != 0)
              throw (runtime_error("Tablename not available"));
  
          //else if((fMode != 0)||(fMode==3))	//TODO make it mode3 + jobID
          //else if((fMode == 1)||(fMode == 2))	//TODO make it mode3 + jobID
          //    throw (runtime_error("Tablename not available"));
          if ((fPmFile.length() > 0) && (2 == getMode()))
          {
              //if(fPmFilePath.length()>0)
              //	aSS << " " << fPmFilePath;
              aSS << " " << fPmFile;
          }
          else if (2 == getMode())
              throw (runtime_error("loadFile [-l ] not available"));
  
      }
  
      if ((fLocFile.length() > 0) && (fLocFile != "STDIN") && (3 == getMode()))
      {
          //Bug 4342 multi-files mode 3 support
          //convert all the spaces into 'commas'
          if (fLocFile.find_first_of(' ') == string::npos)
              aSS << " " << fLocFile;
          else
          {
              std::string aLocFiles = replaceCharInStr(fLocFile, ' ', ',');
              aSS << " " << aLocFiles;
          }
      }
  
      try
      {
          aCmdLine = aSS.str();
      }
      catch (exception&)
      {
          throw runtime_error("getcpImportCmdLine failed");
      }
  
      return aCmdLine;
  }
  
  
  //----------------------------------------------------------------------
  
  //BUG 4205 (part FIX) - need to implement more into it
  bool WECmdArgs::checkForCornerCases()
  {
      //BUG 4210
      this->checkJobIdCase();		//Need to do this before we go further
  
  
      if (fMode == 0)
      {
          if (!fJobId.empty())
          {
              //cout << "Invalid option mode 0 with a Job File" << endl;
              throw (runtime_error("Mode 0 with a Job file option is not valid!!"
                                   "\nTry 'cpimport -h' for more information."));
          }
          else if (!fJobPath.empty())
          {
              cout << "Invalid option mode 0 with a Job Path" << endl;
              throw (runtime_error("Mismatched options"
                                   "\nTry 'cpimport -h' for more information."));
          }
          else if (!fSchema.empty())
          {
              cout << "Invalid option in mode 0 with a schema name" << endl;
              throw (runtime_error("Mismatched options."));
          }
          else if (!fTable.empty())
          {
              cout << "Invalid option in mode 0 with a table name" << endl;
              throw (runtime_error("Mismatched options."));
          }
          else if ((!fPmFilePath.empty()) && (fPmFilePath != "STDIN"))
          {
              cout << "Invalid option -f in Mode 0 with value other than STDIN" << endl;
              throw (runtime_error("Mismatched options."));
          }
  
          if (fSetBufSize)
          {
              cout << "Invalid option -B with Mode 0" << endl;
              throw (runtime_error("Mismatched options."));
          }
          else if (fIOReadBufSize)
          {
              cout << "Invalid option -b with Mode 0" << endl;
              throw (runtime_error("Mismatched options."));
          }
          else if (fMaxErrors >= 0)
          {
              cout << "Invalid option -e with Mode 0" << endl;
              throw (runtime_error("Mismatched options."));
          }
          else if (fConsoleLog)
          {
              cout << "Invalid option -i with Mode 0" << endl;
              throw (runtime_error("Mismatched options."));
          }
          else if (fReadBufSize)
          {
              cout << "Invalid option -c with Mode 0" << endl;
              throw (runtime_error("Mismatched options."));
          }
          else if (fNoOfReadThrds)
          {
              cout << "Invalid option -r with Mode 0" << endl;
              throw (runtime_error("Mismatched options."));
          }
          else if (fNoOfWriteThrds)
          {
              cout << "Invalid option -w with Mode 0" << endl;
              throw (runtime_error("Mismatched options."));
          }
  
          if (fImportDataMode != IMPORT_DATA_TEXT)
          {
              cout << "Invalid option -I with Mode 0" << endl;
              throw (runtime_error("Mismatched options."));
          }
  
      }
  
      if (fMode == 1)
      {
          if (!fJobId.empty())
          {
              if ((!fPmFilePath.empty()) && (fPmFilePath == "STDIN"))
              {
                  // do not do anything.. this is good.
              }
          }
          // Mode 1, if Input Path is existing and input file is not existing
          // it is an error, bce it assumes all the files in directory.
          // In mode 2, we are passing info to cpimport.bin, which will take care
          // of it, as in Mode 3.
          else if ((!fPmFilePath.empty()) && (fPmFile.empty()))
          {
              // assumed since position param is missing
              if ((fLocFile == "STDIN") && (fPmFilePath != "STDIN"))
              {
                  cout << "Invalid options in Mode 1 : option -l " << endl;
                  cout << " or input file position parameter needed" << endl;
                  //cout << "\tOption (-j) should follow with option -l option or "
                  //		"an input file position parameter" << endl;
                  throw (runtime_error("In Mode 1 Error occurred!! "
                                       "\nTry 'cpimport -h' for more information."));
              }
          }
      }
  
      if (fMode == 2)
      {
          if (fPmFile.empty())
              throw(runtime_error("Mode 2 require PM based filename [-l]"
                                  "\nTry 'cpimport -h' for more information."));
  
          if ((fPmFilePath.empty()) && (fPmFile.at(0) != '/'))
              throw(runtime_error("Mode 2 require remote file opts -f and -l or "\
                                  "a fully qualified path for the remote file."
                                  "\nTry 'cpimport -h' for more information."));
          if (!fS3Key.empty())
              throw(runtime_error("Mode 2 & an input file from S3 does not make sense."));
      }
  
      if (fMode == 3)
      {
          if (fPmVec.size())
          {
              cout << "Invalid option -P with Mode 3" << endl;
              throw (runtime_error("Mismatched options."));
          }
  
      }
  
      return true;
  }
  
  //----------------------------------------------------------------------
  
  bool WECmdArgs::str2PmList(std::string& PmList, VecInts& V)
  {
      const int BUFLEN = 512;
      char aBuff[BUFLEN];
  
  
      int aLen = PmList.length();
  
      if (aLen > 0)
      {
          strncpy(aBuff, PmList.c_str(), BUFLEN);
          aBuff[BUFLEN - 1] = 0;
      }
      else
          return false;
  
      char* pTok = strtok(aBuff, ",");
  
      while (pTok != NULL)
      {
          int aPmId = 0;
  
          try
          {
              aPmId = atoi(pTok);
              V.push_back(aPmId);
          }
          catch (exception& ex)
          {
              std::stringstream aErr;
              aErr << "Wrong PM id format : " << ex.what();
              //cout << "Wrong PM id format : "<< ex.what() << endl;
              throw (runtime_error(aErr.str()));
          }
  
          pTok = strtok(NULL, ",");
      }
  
      return true;
  }
  
  //----------------------------------------------------------------------
  
  void WECmdArgs::usage()
  {
      cout << "Simple usage using positional parameters (no XML job file):\n";
      cout << "\tcpimport dbName tblName [loadFile] [-h] [-m mode]\n";
      cout << "\t\t [-f path] [-d debugLevel] [-c readbufSize] [-b readBufs] \n";
      cout << "\t\t [-r readers] [-j JobID] [-e maxErrs] [-B libBufSize] [-w parsers]\n";
      cout << "\t\t [-s c] [-E enclosedChar] [-C escapeChar] [-n NullOption]\n";
      cout << "\t\t [-q batchQty] [-p jobPath] [-P list of PMs] [-S] [-i] [-v verbose]\n";
      cout << "\t\t [-I binaryOpt] [-T timeZone]\n";
  
  
      cout << "Traditional usage without positional parameters (XML job file required):\n";
      cout << "\tcpimport -j jobID\n";
      cout << "\t\t [-h] [-m mode] [-r readers] [-w parsers] [-s c] [-f path]\n";
      cout << "\t\t [-b readBufs] [-p path] [-c readBufSize] [-e maxErrs] [-B libBufSize]\n";
      cout << "\t\t [-n NullOption] [-E encloseChar] [-C escapeChar] [-i] [-v verbose]\n";
      cout << "\t\t [-d debugLevel] [-q batchQty] [-l loadFile] [-P list of PMs] [-S]\n";
      cout << "\t\t [-I binaryOpt] [-T timeZone]\n";
  
      cout << "\n\nPositional parameters:\n";
      cout << "\tdbName     Name of the database to load\n";
      cout << "\ttblName    Name of table to load\n";
      cout << "\tloadFile   Optional input file name in current directory,\n";
      cout << "\t\t\tunless a fully qualified name is given.\n";
      cout << "\t\t\tIf not given, input read from STDIN.\n";
  
      cout << "\n\nOptions:\n"
           << "\t-b\tNumber of read buffers\n"
           << "\t-c\tApplication read buffer size(in bytes)\n"
           << "\t-d\tPrint different level(1-3) debug message\n"
           << "\t-e\tMax number of allowable error per table per PM\n"
           << "\t-f\tData file directory path.\n"
           << "\t\t\tDefault is current working directory.\n"
           << "\t\t\tIn Mode 1, -f represents the local input file path.\n"
           << "\t\t\tIn Mode 2, -f represents the PM based input file path.\n"
           << "\t\t\tIn Mode 3, -f represents the local input file path.\n"
           << "\t-l\tName of import file to be loaded, relative to -f path,\n"
           << "\t-h\tPrint this message.\n"
           << "\t-q\tBatch Quantity, Number of rows distributed per batch in Mode 1\n"
           << "\t-i\tPrint extended info to console in Mode 3.\n"
           << "\t-j\tJob ID. In simple usage, default is the table OID.\n"
           << "\t\t\tunless a fully qualified input file name is given.\n"
           << "\t-n\tNullOption (0-treat the string NULL as data (default);\n"
           << "\t\t\t1-treat the string NULL as a NULL value)\n"
           << "\t-p\tPath for XML job description file.\n"
           << "\t-r\tNumber of readers.\n"
           << "\t-s\t'c' is the delimiter between column values.\n"
           << "\t-B\tI/O library read buffer size (in bytes)\n"
           << "\t-w\tNumber of parsers.\n"
           << "\t-E\tEnclosed by character if field values are enclosed.\n"
           << "\t-C\tEscape character used in conjunction with 'enclosed by'\n"
           << "\t\t\tcharacter, or as part of NULL escape sequence ('\\N');\n"
           << "\t\t\tdefault is '\\'\n"
           << "\t-I\tImport binary data; how to treat NULL values:\n"
           << "\t\t\t1 - import NULL values\n"
           << "\t\t\t2 - saturate NULL values\n"
           << "\t-P\tList of PMs ex: -P 1,2,3. Default is all PMs.\n"
           << "\t-S\tTreat string truncations as errors.\n"
           << "\t-m\tmode\n"
           << "\t\t\t1 - rows will be loaded in a distributed manner across PMs.\n"
           << "\t\t\t2 - PM based input files loaded onto their respective PM.\n"
           << "\t\t\t3 - input files will be loaded on the local PM.\n"
           << "\t-T\tTimezone used for TIMESTAMP datatype.\n"
           << "\t\tPossible values: \"SYSTEM\" (default)\n"
           << "\t\t               : Offset in the form +/-HH:MM\n"
           << "\t-y\tS3 Authentication Key (for S3 imports)\n"
           << "\t-K\tS3 Authentication Secret (for S3 imports)\n"
           << "\t-t\tS3 Bucket (for S3 imports)\n"
           << "\t-H\tS3 Hostname (for S3 imports, Amazon's S3 default)\n"
           << "\t-g\tS3 Region (for S3 imports)\n"
           << "\t-L\tDirectory for the output .err and .bad files.\n"
           << "\t\tDefault is " << string(MCSLOGDIR);
  
  
      cout << "\nExample1: Traditional usage\n"
           << "\tcpimport -j 1234";
      cout << "\nExample2: Some column values are enclosed within double quotes.\n"
           << "\tcpimport -j 3000 -E '\"'";
      cout << "\nExample3: Import a nation table without a Job XML file\n"
           << "\tcpimport -j 301 tpch nation nation.tbl";
      cout << "\nExample4: Import a nation table to all PMs in Mode 1\n"
           << "\tcpimport -m 1 tpch nation nation.tbl";
      cout << "\nExample4: Import a nation table to only PM1 and PM2 in Mode 1\n"
           << "\tcpimport -m 1 -P 1,2 tpch nation nation.tbl";
      cout << "\nExample5: Import nation.tbl from PMs to nation table in Mode 2\n"
           << "\tcpimport -m 2 tpch nation -f /var/lib/columnstore/data/bulk/data/import/ -l nation.tbl";
      cout << "\nExample6: Import nation.tbl in mode 3\n"
           << "\tcpimport -m 3 tpch nation nation.tbl\n\n";
  
  
      exit(1);
  }
  
  
  //-----------------------------------------------------------------------------
  
  void WECmdArgs::parseCmdLineArgs(int argc, char** argv)
  {
      int aCh;
      std::string importPath;
      bool aJobType = false;
  
  
      if (argc > 0)
          fPrgmName = string(MCSBINDIR) + "/" + "cpimport.bin"; //argv[0] is splitter but we need cpimport
  
      while ((aCh = getopt(argc, argv,
                           "d:j:w:s:v:l:r:b:e:B:f:q:ihm:E:C:P:I:n:p:c:ST:Ny:K:t:H:g:U:L:"))
              != EOF)
      {
          switch (aCh)
          {
              case 'm':
              {
                  fArgMode = atoi(optarg);
  
                  //cout << "Mode level set to " << fMode << endl;
                  if ((fArgMode > -1) && (fArgMode <= 3)) {}
                  else
                      throw runtime_error("Wrong Mode level");
  
                  break;
              }
  
              case 'B':
              {
                  errno = 0;
                  long lValue = strtol(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 1) || (lValue > INT_MAX))
                      throw runtime_error("Option -B is invalid or out of range");
  
                  fSetBufSize = lValue;
                  break;
              }
  
              case 'b':
              {
                  errno = 0;
                  long lValue = strtol(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 1) || (lValue > INT_MAX))
                      throw runtime_error("Option -b is invalid or out of range");
  
                  fIOReadBufSize = lValue;
                  break;
              }
  
              case 'e':
              {
                  errno = 0;
                  long lValue = strtol(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 0) || (lValue > INT_MAX))
                      throw runtime_error("Option -e is invalid or out of range");
  
                  fMaxErrors = lValue;
                  break;
              }
  
              case 'i':
              {
                  fConsoleLog = true;
                  break;
              }
  
              case 'c':
              {
                  errno = 0;
                  long lValue = strtol(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 1) || (lValue > INT_MAX))
                      throw runtime_error("Option -c is invalid or out of range");
  
                  fReadBufSize = lValue;
                  break;
              }
  
              case 'j': // -j: jobID
              {
                  errno = 0;
                  long lValue = strtol(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 0) || (lValue > INT_MAX))
                      throw runtime_error("Option -j is invalid or out of range");
  
                  fJobId = optarg;
                  fOrigJobId = fJobId;	// in case if we need to split it.
  
                  if (0 == fJobId.length()) throw runtime_error("Wrong JobID Value");
  
                  aJobType = true;
                  break;
              }
  
              case 'v': // verbose
              {
                  string aVerbLen = optarg;
                  fVerbose = aVerbLen.length();
                  fDebugLvl = fVerbose;
                  break;
              }
  
              case 'd': // -d debug
              {
                  errno = 0;
                  long lValue = strtol(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 1) || (lValue > INT_MAX))
                      throw runtime_error("Option -d is invalid or out of range");
  
                  fDebugLvl = lValue;
  
                  if (fDebugLvl > 0 && fDebugLvl <= 3)
                  {
                      cout << "\nDebug level set to " << fDebugLvl << endl;
                  }
                  else
                  {
                      throw runtime_error("Wrong Debug level");
                  }
  
                  break;
              }
  
              case 'r': // -r: num read threads
              {
                  errno = 0;
                  long lValue = strtol(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 1) || (lValue > INT_MAX))
                      throw runtime_error("Option -r is invalid or out of range");
  
                  fNoOfReadThrds = lValue;
                  break;
              }
  
              case 'w': // -w: num parse threads
              {
                  errno = 0;
                  long lValue = strtol(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 1) || (lValue > INT_MAX))
                      throw runtime_error("Option -w is invalid or out of range");
  
                  fNoOfWriteThrds = lValue;
                  break;
              }
  
              case 's': // -s: column delimiter
              {
                  if (!strcmp(optarg, "\\t"))
                  {
                      fColDelim = '\t';
  
                      if (fDebugLvl) cout << "Column delimiter : " << "\\t" << endl;
                  }
                  else
                  {
                      fColDelim = optarg[0];
  
                      if (fDebugLvl) cout << "Column delimiter : " << fColDelim << endl;
                  }
  
                  break;
              }
  
              case 'l': // -l: if JobId (-j), it can be input file
              {
                  fPmFile = optarg;
  
                  if (0 == fPmFile.length()) throw runtime_error("Wrong local filename");
  
                  break;
              }
  
              case 'f': // -f: import file path
              {
                  fPmFilePath = optarg;
                  break;
              }
  
              case 'n': // -n: treat "NULL" as null
              {
                  // default is 0, ie it is equal to not giving this option
                  int nullStringMode = atoi(optarg);
  
                  if ((nullStringMode != 0) && (nullStringMode != 1))
                  {
                      throw (runtime_error(
                                 "Invalid NULL option; value can be 0 or 1"));
                  }
  
                  if (nullStringMode)
                      fNullStrMode = true;
                  else
                      fNullStrMode = false; // This is default
  
                  break;
              }
  
              case 'P': // -p: list of PM's
              {
                  try
                  {
                      std::string aPmList = optarg;
  
                      if (!str2PmList(aPmList, fPmVec))
                          throw(runtime_error("PM list is wrong"));
                  }
                  catch (runtime_error& ex)
                  {
                      throw (ex);
                  }
  
                  break;
              }
  
              case 'p':
              {
                  fJobPath = optarg;
                  break;
              }
  
              case 'E': // -E: enclosed by char
              {
                  fEnclosedChar = optarg[0];
                  //cout << "Enclosed by Character : " << optarg[0] << endl;
                  break;
              }
  
              case 'C': // -C: enclosed escape char
              {
                  fEscChar = optarg[0];
                  //cout << "Escape Character  : " << optarg[0] << endl;
                  break;
              }
  
              case 'h': // -h: help
              {
                  //usage(); // will exit(1) here
                  fHelp = true;
                  break;
              }
  
              case 'I': // -I: binary mode (null handling)
              {
                  // default is text mode, unless -I option is specified
                  int binaryMode = atoi(optarg);
  
                  if (binaryMode == 1)
                  {
                      fImportDataMode = IMPORT_DATA_BIN_ACCEPT_NULL;
                  }
                  else if (binaryMode == 2)
                  {
                      fImportDataMode = IMPORT_DATA_BIN_SAT_NULL;
                  }
                  else
                  {
                      throw (runtime_error(
                                 "Invalid Binary mode; value can be 1 or 2"));
                  }
  
                  break;
              }
  
              case 'S': // -S: Treat string truncations as errors
              {
                  setTruncationAsError(true);
                  //cout << "TruncationAsError  : true" << endl;
                  break;
              }
  
              case 'T':
              {
                  std::string timeZone = optarg;
                  long offset;
  
                  if (timeZone != "SYSTEM" && dataconvert::timeZoneToOffset(timeZone.c_str(), timeZone.size(), &offset))
                  {
                      throw (runtime_error(
                                 "Value for option -T is invalid"));
                  }
  
                  fTimeZone = timeZone;
                  break;
              }
  
              case 'q': // -q: batch quantity - default value is 10000
              {
                  errno = 0;
                  long long lValue = strtoll(optarg, 0, 10);
  
                  if ((errno != 0) || (lValue < 1) || (lValue > UINT_MAX))
                      throw runtime_error("Option -q is invalid or out of range");
  
                  fBatchQty = lValue;
  
                  if (fBatchQty < 10000) fBatchQty = 10000;
                  else if (fBatchQty > 100000) fBatchQty = 10000;
  
                  break;
              }
  
              case 'N': //-N no console output
              {
                  fConsoleOutput = false;
                  break;
              }
  
              case 'y': //-y S3 Key
              {
                  fS3Key = optarg;
                  break;
              }
  
              case 'K': //-K S3 Secret
              {
                  fS3Secret = optarg;
                  break;
              }
  
              case 'H': //-H S3 Host
              {
                  fS3Host = optarg;
                  break;
              }
  
              case 't': //-t S3 bucket
              {
                  fS3Bucket = optarg;
                  break;
              }
  
              case 'g': //-g S3 Region
              {
                  fS3Region = optarg;
                  break;
              }
  
              case 'U': //-U username of the files owner
              {
                  fUsername = optarg;
                  break;
              }
  
              case 'L': // -L set the output location of .bad/.err files
              {
                  fErrorDir = optarg;
                  break;
              }
  
              default:
              {
                  std::string aErr = std::string("Unknown command line option ") + std::to_string(aCh);
                  //cout << "Unknown command line option " << aCh << endl;
                  throw (runtime_error(aErr));
              }
          }
      }
  
      if (fHelp) usage();	//BUG 4210
  
      if (fArgMode != -1) fMode = fArgMode;	//BUG 4210
  
      std::string bulkRootPath = getBulkRootDir();
  
      checkForBulkLogDir(bulkRootPath);
  
      if (2 == fArgMode && fPmFilePath.empty())
          throw runtime_error("-f option is mandatory with mode 2.");
  
      if (aJobType)
      {
          if (0 == fArgMode) throw runtime_error("Incompatible mode and option types");
  
          if (optind < argc)
          {
              fSchema = argv[optind]; // 1st pos parm
              optind++;
  
              if (optind < argc)
              {
                  fTable = argv[optind]; // 2nd pos parm
                  optind++;
              }
              else
              {
                  // if schema is there, table name should be there
                  throw runtime_error("No table name specified with schema.");
              }
  
              if (optind < argc) // see if input file name is given
              {
                  // 3rd pos parm
                  fLocFile = argv[optind];
  
                  if ((fLocFile.at(0) != '/') && (fLocFile != "STDIN"))
                  {
                      std::string aTmp = fLocFile;
  
                      // BUG 4379 -f given? use that
                      if ((!fPmFilePath.empty()) && (fMode == 1))
                          fLocFile = fPmFilePath + "/" + aTmp;
                      else if (fPmFilePath.empty())
                          fLocFile = bulkRootPath + "/data/import/" + aTmp;
                  }
              }
              else
              {
                  if (!fPmFile.empty())
                      fLocFile = fPmFile;
                  //BUG 4186
                  //else  // else take it from the jobxml file
                  //	fLocFile = "STDIN";
                  //Historically cpimport works with jobfile as
                  // -l <fileName> && -f <filePath>   or
                  // -fSTDIN as the stdin, it will override colxml loadfile entry
                  // if -fSTDIN is not provided get i/p file from jobfile
                  else if ((!fPmFilePath.empty()) && (fPmFilePath == "STDIN"))
                      fLocFile = "STDIN";
  
                  // else take it from the jobxml file
              }
  
              if ((fSchema.length() > 0) && (fTable.length() > 0) && (fLocFile.length() > 0))
                  fJobLogOnly = true;
          }
          else
          {
              if (!fPmFile.empty())
              {
                  fLocFile = fPmFile;
  
                  if (!fPmFilePath.empty())
                  {
                      if (fPmFilePath == "STDIN")
                      {
                          throw runtime_error("Conflicting options -l and -fSTDIN");
                      }
                      else
                      {
                          std::string aTmp = fLocFile;
  
                          if ((!fPmFilePath.empty()) && (fMode == 1)) //BUG 4379 -f given? use that
                              fLocFile = fPmFilePath + "/" + aTmp;
                          else if (!fPmFilePath.empty())
                              fLocFile = bulkRootPath + "/data/import/" + aTmp;
                      }
  
                  }
  
                  if ((fLocFile.at(0) != '/') && (fLocFile != "STDIN") && (fPmFilePath.empty()))
                  {
                      std::string aTmp = fLocFile;
                      fLocFile = bulkRootPath + "/data/import/" + aTmp;
                  }
              }
              //BUG 4186
              //else
              //	fLocFile = "STDIN";
              //Historically cpimport works with jobfile as
              // -l <fileName> && -f <filePath>   or
              // -fSTDIN as the stdin, it will override colxml loadfile entry
              // if -fSTDIN is not provided get i/p file from jobfile
              else if ((!fPmFilePath.empty()) && (fPmFilePath == "STDIN"))
                  fLocFile = "STDIN";
  
              // else take it from the jobxml file
          }
  
      }
      // Get positional arguments, User can provide:
      // 1. no positional parameters	- Mode 0 & stdin
      // 2. Two positional parameters (schema and table names) - Mode 1/2, stdin
      // 3. Three positional parameters (schema, table, and import file name)
      else if (optind < argc) // see if db schema name is given
      {
          if (fArgMode == 0)
          {
              //added the code as per BUG 4245
              if (!fPmFilePath.empty())
              {
                  fLocFile = fPmFilePath;
  
                  if (fLocFile != "STDIN")
                      throw(runtime_error("ERROR: In Mode 0, -f option can only have value STDIN"));
              }
              else
              {
                  fLocFile = argv[optind];
                  optind++;
              }
  
              if (optind < argc) //dest filename provided
              {
                  fPmFile = argv[optind];
  
                  if ((fPmFile.at(0) != '/') && (fS3Key.empty()))
                  {
                      std::string aTmp = fPmFile;
                      fPmFile = bulkRootPath + "/data/import/" + aTmp;
                  }
              }
              else // no dest filename
              {
                  if (fLocFile == "STDIN")
                      throw(runtime_error("ERROR: Destination file name required!!"));
  
                  if (fLocFile.at(0) == '/')	//local FQ-filename,parse out filename
                      fPmFile = getFileNameFromPath(fLocFile);
                  else
                      fPmFile = fLocFile;
  
                  if ((fPmFile.at(0) != '/') && (fS3Key.empty()))	//should be true all the time
                  {
                      std::string aTmp = fPmFile;
                      fPmFile = bulkRootPath + "/data/import/" + aTmp;
                  }
              }
  
              /* commented out for BUG 4245
              if(fPmFilePath.empty())
              	fLocFile = argv[optind];
              else
              	fLocFile = fPmFilePath +"/"+ argv[optind];
  
              if (fPmFile.empty()) //BUG 4200
              {
              	//if(fLocFile.at(0)== '/')
              	//	fPmFile = fLocFile;
              	//else
              	if(fLocFile.at(0)!='/')
              		fPmFile = bulkRootPath + "/data/import/"+ fLocFile;
              }
              else
              {
              	if(fPmFile.at(0)!='/')
              	{
              		std::string aTmp = fPmFile;
              		fPmFile = bulkRootPath + "/data/import/"+aTmp;
              	}
              }
              */
          }
          else
              fSchema = argv[optind]; // 1st pos parm
  
          optind++;
  
          if (optind < argc) // see if table name is given
          {
              fTable = argv[optind]; // 2nd pos parm
              optind++;
  
              if (optind < argc) // see if input file name is given
              {
                  // 3rd pos parm
                  fLocFile = argv[optind];
  
                  //BUG 4379 if -f option given we need to use that path,
                  //over riding bug 4231. look at the code below
                  //BUG 4231 - This bug over writes 4199 and commenting out changes
                  //BUG 4199
                  //Path not provided, not fully qualified, Look in import dir
                  //if((fLocFile.at(0)!= '/')&&(fLocFile != "STDIN"))
                  //{
                  //	std::string aTmp = fLocFile;
                  //	fLocFile = bulkRootPath + "/data/import/"+ aTmp;
                  //}
                  //BUG 4379 if -f option given we need to use that path
                  if ((fLocFile.at(0) != '/') && (fLocFile != "STDIN"))
                  {
                      std::string aTmp = fLocFile;
  
                      //if -f given? use that otherwise just go ahead with CWD
                      if ((!fPmFilePath.empty()) && (fMode == 1))
                          fLocFile = fPmFilePath + "/" + aTmp;
  
                      // TODO - if -f option is given and a list of files are
                      // are provided, we need to be able to import all that.
                  }
  
  
              }
              else
              {
                  if (fPmFile.length() > 0)
                  {
                      // BUG 4210
                      //if (fPmFilePath.length() > 0)
                      //{
                      //	fLocFile = fPmFilePath +"/"+ fPmFile;
                      //}
                      //else
                      if (fPmFilePath.empty())
                      {
                          //NOTE - un-commenting with an if statement for Mode 2
                          //BUG 4231 makes it comment out the below changes,
                          //This will not change even though directly, to be
                          //on safer side, we should take out this too.
                          //check path fully qualified? then set as data import
                          if (2 == fArgMode)
                          {
                              //BUG 4342
                              if ((fPmFile.at(0) != '/') && (fS3Key.empty()))
                              {
                                  std::string aTmp = fPmFile;
                                  fPmFile = PrepMode2ListOfFiles(aTmp);
                              }
                              else
                              {
                                  if (fPmFile.find_first_of(' ') != string::npos)
                                  {
                                      std::string aPmFiles = replaceCharInStr(fPmFile, ' ', ',');
                                      fPmFile = aPmFiles;
                                  }
                              }
                          }
  
                          fLocFile = fPmFile;
                      }
                  }
                  else
                  {
                      fLocFile = "STDIN";
                  }
  
                  //cout << "LocFile set as stdin" << endl;
              }
          }
          else
          {
              // If Mode is not 0 and table name is a required argument
              if (fArgMode != 0)
                  throw(runtime_error("No table name specified with schema."));
          }
  
      }
      else
      {
          // for testing we are allowing data from stdin even with Mode 0
          // that is without LocFileName
          if (0 == fArgMode)
          {
              fLocFile = "STDIN";	//cout << "LocFile set as stdin" << endl;
          }
          else
          {
              // If Mode 0, LocFileName is reqd and otherwies Schema is required
              throw (runtime_error("No schema or local filename specified."));
          }
      }
  
      /* check for all-or-nothing cmdline args to enable S3 import */
      int s3Tmp = (fS3Key.empty() ? 0 : 1) + (fS3Bucket.empty() ? 0 : 1) +
          (fS3Secret.empty() ? 0 : 1) + (fS3Region.empty() ? 0 : 1);
      if (s3Tmp != 0 && s3Tmp != 4)
          throw runtime_error("The access key, secret, bucket, and region are all required to import from S3");
  }
  
  std::string WECmdArgs::getJobFileName()
  {
      std::ostringstream aSS;
      string aJobIdFileName;
  
      if (fJobId.length() > 0)
      {
          if (fJobPath.length() > 0)
              aSS << fJobPath;
          else
          {
              fJobPath = config::Config::makeConfig()->getConfig("WriteEngine",
                         "BulkRoot") + "/Job";
              aSS << fJobPath;
          }
  
          aSS << "/Job_" << fJobId << ".xml";
          aJobIdFileName = aSS.str();
      }
  
      return aJobIdFileName;
  }
  
  bool WECmdArgs::getPmStatus(int Id)
  {
      // if no PMID's provided on cmdline, return true;
      if (0 == fPmVec.size()) return true;
  
      VecInts::iterator aIt = fPmVec.begin();
  
      while (aIt != fPmVec.end())
      {
          if (*aIt == static_cast<unsigned int>(Id))
              return true;
  
          ++aIt;
      }
  
      return false;
  }
  
  
  //------------------------------------------------------------------------------
  // It is a recursive call.
  std::string WECmdArgs::getBrmRptFileName()
  {
      if (!fBrmRptFile.empty())
          return fBrmRptFile;
  
      string brmRptFileName = getTmpFileDir();
  
      if (!brmRptFileName.empty())
      {
          fTmpFileDir = brmRptFileName;
          char aBuff[64];
          time_t aTime;
          struct tm pTm;
          time(&aTime);
          localtime_r(&aTime, &pTm);
  
          // BUG 4424
          //			M   D   H   M   S
          snprintf(aBuff, sizeof(aBuff), "/BrmRpt%02d%02d%02d%02d%02d%d.rpt",
                   pTm.tm_mon, pTm.tm_mday, pTm.tm_hour,
                   pTm.tm_min, pTm.tm_sec, getpid());
          brmRptFileName += aBuff;
      }
      else
      {
          //cout << "ERROR: Could not find TempFileDir in Columnstore.xml" << endl;
          throw (runtime_error("Could not find TempFileDir in Columnstore.xml"));
      }
  
      setBrmRptFileName(brmRptFileName);
  
      return brmRptFileName;
  
  }
  //------------------------------------------------------------------------------
  
  void WECmdArgs::addJobFilesToVector(std::string& JobName)
  {
      //if((!fSchema.empty())&&(!fTable.empty())&&(!fLocFile.empty())) return;
  
      WEXmlgetter aXmlGetter(JobName);
      vector<string> aSections;
      aSections.push_back("BulkJob");
      aSections.push_back("Schema");
      aSections.push_back("Table");
  
      //BUG 4163
      typedef std::vector<string> TableVec;
      TableVec aTableVec;
      aXmlGetter.getConfig(aSections[1], aSections[2], aTableVec);
      setMultiTableCount(aTableVec.size());
  
      if (getMultiTableCount() > 1)
      {
          splitConfigFilePerTable(JobName, aTableVec.size());
      }
      else
      {
          fVecJobFiles.push_back(JobName);
      }
  
  }
  
  //------------------------------------------------------------------------------
  // Set the schema, table, and loadfile name from the xml job file.
  // If running in binary mode, we also get the list of columns for the table,
  // so that we can determine the exact fixed record length of the incoming data.
  //------------------------------------------------------------------------------
  void WECmdArgs::setSchemaAndTableFromJobFile(std::string& JobName)
  {
      if (((fVecJobFiles.size() == 1) && (!fSchema.empty()) &&
              (!fTable.empty()) && (!fLocFile.empty()))  &&
              (fImportDataMode == IMPORT_DATA_TEXT)) return;
  
      WEXmlgetter aXmlGetter(JobName);
      vector<string> aSections;
      aSections.push_back("BulkJob");
      aSections.push_back("Schema");
      aSections.push_back("Table");
  
      // Reset the fSchema, fTable, and FLocFile
      if ((fVecJobFiles.size() > 1) ||
              (fSchema.empty()) || (fTable.empty()) || (fLocFile.empty()))
      {
          std::string aSchemaTable;
          std::string aInputFile;
  
          aSchemaTable = aXmlGetter.getAttribute(aSections, "tblName");
  
          if (getDebugLvl() > 1) cout << "schema.table = " << aSchemaTable << endl;
  
          aInputFile = aXmlGetter.getAttribute(aSections, "loadName");
  
          if (getDebugLvl() > 1) cout << "xml::InputFile = " << aInputFile << endl;
  
          if (aSchemaTable.length() > 0)
          {
              char aSchema[64];
              char aTable[64];
              int aRet = aSchemaTable.find('.');
  
              if (aRet > 0)
              {
                  int aLen = aSchemaTable.copy(aSchema, aRet);
  
                  if (getDebugLvl() > 1) cout << "Schema: " << aSchema << endl;
  
                  aSchema[aLen] = 0;
  
                  if (fSchema.empty()) fSchema = aSchema;
  
                  aLen = aSchemaTable.copy(aTable, aSchemaTable.length(), aRet + 1 );
                  aTable[aLen] = 0;
  
                  if (getDebugLvl() > 1) cout << "Table: " << aTable << endl;
  
                  fTable = aTable;
              }
              else
                  throw runtime_error(
                      "JobFile ERROR: Can't get Schema and Table Name");
          }
          else
          {
              throw runtime_error(
                  "JobFile ERROR: Can't get Schema and Table Name");
          }
  
          if ((fLocFile.empty()) && (!aInputFile.empty()))
          {
              string bulkRootPath = config::Config::makeConfig()->getConfig(
                                        "WriteEngine", "BulkRoot");
  
              if (aInputFile.at(0) == '/')
                  fLocFile = aInputFile;
              else if ((!fPmFilePath.empty()) && (fMode == 1))
                  fLocFile = fPmFilePath + "/" + aInputFile;
              else if ((!bulkRootPath.empty()) && (fPmFilePath.empty()))
                  fLocFile = bulkRootPath + "/data/import/" + aInputFile;
              else
                  fLocFile = aInputFile;
  
              if (fArgMode == 2) fPmFile = fLocFile;
          }
  
          if (getDebugLvl() > 1) cout << "schema = " << fSchema << endl;
  
          if (getDebugLvl() > 1) cout << "TableName = " << fTable << endl;
  
          if (getDebugLvl() > 1) cout << "Input File = " << fLocFile << endl;
      }
  
      // Reset the list of columns we will be importing from the input data
      fColFldsFromJobFile.clear();
  
      if (fImportDataMode != IMPORT_DATA_TEXT)
      {
          aSections.push_back("Column");
          aXmlGetter.getAttributeListForAllChildren(
              aSections, "colName", fColFldsFromJobFile);
      }
  }
  
  //------------------------------------------------------------------------------
  void WECmdArgs::checkJobIdCase()
  {
      if ((fJobId.empty()) || (fJobLogOnly) || (fMode == 3) || (fMode == 0)) return;
  
      if (fJobPath.empty())
      {
          string bulkRootPath = config::Config::makeConfig()->getConfig(
                                    "WriteEngine", "BulkRoot");
          //cout << "checkJobIdCase::BulkRoot: " << bulkRootPath << endl;
  
          if (!bulkRootPath.empty())
              fJobPath = bulkRootPath + "/job";
          else
              throw runtime_error("Config Error: BulkRoot not found in Columnstore.xml");
      }
  
      char aBuff[256];
  
      if (!fJobPath.empty())
          snprintf(aBuff, sizeof(aBuff), "%s/Job_%s.xml", fJobPath.c_str(),
                   fJobId.c_str());
      else	// for time being
          snprintf(aBuff, sizeof(aBuff), "/var/log/mariadb/columnstore/data/bulk/job/Job_%s.xml",
                   fJobId.c_str());
  
      std::string aJobFileName(aBuff);
  
      //cout << "checkJobIdCase::aJobFileName: " << aJobFileName << endl;
  
  
      //BUG 4171
      addJobFilesToVector(aJobFileName);
  
      aJobFileName =  fVecJobFiles[0];
      setSchemaAndTableFromJobFile(aJobFileName);
      setEnclByAndEscCharFromJobFile(aJobFileName);
  
  }
  
  //------------------------------------------------------------------------------
  
  std::string WECmdArgs::getTmpFileDir()
  {
      if (!fTmpFileDir.empty()) return fTmpFileDir;
  
      fTmpFileDir = startup::StartUp::tmpDir();
  
      if (fTmpFileDir.empty())
          throw( runtime_error("Config ERROR: TmpFileDir not found!!"));
      else
          return fTmpFileDir;
  }
  
  //------------------------------------------------------------------------------
  
  std::string WECmdArgs::getBulkRootDir()
  {
      if (!fBulkRoot.empty()) return fBulkRoot;
  
      fBulkRoot = config::Config::makeConfig()->getConfig("WriteEngine",
                  "BulkRoot");
  
      if (fBulkRoot.empty())
          throw( runtime_error("Config ERROR: <BulkRoot> not found!!"));
      else
          return fBulkRoot;
  }
  
  //------------------------------------------------------------------------------
  
  unsigned int WECmdArgs::getBatchQuantity()
  {
      return (fBatchQty >= 10000) ? fBatchQty : 10000;	//default Batch Qty is 10000
  }
  
  //------------------------------------------------------------------------------
  
  void WECmdArgs::setEnclByAndEscCharFromJobFile(std::string& JobName)
  {
      if (fEnclosedChar == 0)	// check anything in Jobxml file
      {
          WEXmlgetter aXmlGetter(JobName);
          vector<string> aSections;
          aSections.push_back("BulkJob");
          aSections.push_back("EnclosedByChar");
  
          try
          {
              //std::string aTable = aXmlGetter.getConfig(aSection, aElement);
              std::string aEnclosedBy = aXmlGetter.getValue(aSections);
  
              if (getDebugLvl() > 1)cout << "aEncloseBy = " << aEnclosedBy << endl;
  
              if (!aEnclosedBy.empty())
              {
                  fEnclosedChar = aEnclosedBy.at(0);
              }
          }
          catch (std::runtime_error&)
          {
              // do not do anything
          }
      }
  
      if (fEscChar == 0)	// check anything in Jobxml file
      {
          WEXmlgetter aXmlGetter(JobName);
          vector<string> aSections;
          aSections.push_back("BulkJob");
          aSections.push_back("EscapeChar");
  
          try
          {
              //std::string aTable = aXmlGetter.getConfig(aSection, aElement);
              std::string aEscChar = aXmlGetter.getValue(aSections);
  
              if (getDebugLvl() > 1) cout << "aEscapeChar = " << aEscChar << endl;
  
              if (!aEscChar.empty())
              {
                  fEscChar = aEscChar.at(0);
              }
          }
          catch (std::runtime_error&)
          {
              // do not do anything
          }
      }
  
  }
  
  //------------------------------------------------------------------------------
  std::string WECmdArgs::getFileNameFromPath(const std::string& Path) const
  {
      char aBuff[64];
      int iDx = Path.find_last_of('/');
      iDx++;		// compensate for the forward slash
      int aCx = Path.size() - iDx;
      Path.copy(aBuff, aCx, iDx);
      aBuff[aCx] = 0;
      return aBuff;
  }
  
  //------------------------------------------------------------------------------
  std::string WECmdArgs::getModuleID()
  {
      oam::Oam oam;
      oam::oamModuleInfo_t sModInfo;
      std::string sModuleID;
      char szModuleIDandPID[64];
      int nModuleNumber;
  
      try
      {
          sModInfo = oam.getModuleInfo();
          sModuleID = boost::get < 1 > (sModInfo);
          nModuleNumber = boost::get < 2 > (sModInfo);
          snprintf(szModuleIDandPID, sizeof(szModuleIDandPID), "%s%d-%d",
                   sModuleID.c_str(), nModuleNumber, getpid());
          sModuleID = szModuleIDandPID;
      }
      catch (exception&)
      {
          sModuleID = "unknown";
      }
  
      return sModuleID;
  }
  
  //------------------------------------------------------------------------------
  
  
  void WECmdArgs::splitConfigFilePerTable(std::string& ConfigName, int tblCount)
  {
      std::string aOpenTag = "<Table ";
      std::string aCloseTag = "</Table>";
      std::string aCloseSchemaTag = "</Schema>";
  
      std::vector<std::ofstream*> aVecFiles;
  
      //std::vector<std::string> aVecConfigs;
      for (int aIdx = 1; aIdx <= tblCount; aIdx++)
      {
          char aConfName[128];
          snprintf(aConfName, sizeof(aConfName), "%s_%d.xml", ConfigName.c_str(), aIdx);
          //aVecConfigs.push_back(aConfName);
          fVecJobFiles.push_back(aConfName);
          std::ofstream* pCopy = new std::ofstream;
          //pCopy->open(aConfName, std::ios_base::app);
          pCopy->open(aConfName);
          aVecFiles.push_back(pCopy);
      }
  
  
      std::ifstream aMaster;
      aMaster.open(ConfigName.c_str());
  
      if (aMaster.is_open())
      {
          char aBuff[256];
          int aTblNo = 0;
          size_t aStrPos = std::string::npos;
          bool aOpenFound = false;
          bool aCloseFound = false;
  
          while (!aMaster.eof())
          {
              aMaster.getline(aBuff, sizeof(aBuff) - 1);
              unsigned int aLen = aMaster.gcount();
  
              if ((aLen < (sizeof(aBuff) - 2)) && (aLen > 0))
              {
                  aBuff[aLen - 1] = '\n';
                  aBuff[aLen] = 0;
                  string aData = aBuff;
                  //cout << "Data Read " << aBuff;
  
                  if (!aOpenFound)
                  {
                      aStrPos = aData.find(aOpenTag);
  
                      if (aStrPos != std::string::npos)
                      {
                          aOpenFound = true;
                          aTblNo++;
                          write2ConfigFiles(aVecFiles, aBuff, aTblNo);
                      }
                      else
                      {
                          if ((!aOpenFound) && (aCloseFound))
                          {
                              aStrPos = aData.find(aCloseSchemaTag);
  
                              if (aStrPos != std::string::npos)
                              {
                                  aOpenFound = false;
                                  aCloseFound = false;
                                  aTblNo = 0;
                              }
                          }
  
                          write2ConfigFiles(aVecFiles, aBuff, aTblNo);
                      }
                  }
                  else
                  {
                      aStrPos = aData.find(aCloseTag);
  
                      if (aStrPos != std::string::npos)
                      {
                          aOpenFound = false;
                          aCloseFound = true;
                          write2ConfigFiles(aVecFiles, aBuff, aTblNo);
                      }
                      else
                      {
                          write2ConfigFiles(aVecFiles, aBuff, aTblNo);
                      }
                  }
              }
          }//while Master.eof
      }
      else
      {
          throw runtime_error("Could not open Job Config file");
      }
  
  
      for (unsigned int Idx = 0; Idx < aVecFiles.size(); Idx++)
      {
          aVecFiles[Idx]->close();
          delete aVecFiles[Idx];
      }
  
      aVecFiles.clear();
  
  }
  
  //------------------------------------------------------------------------------
  
  void WECmdArgs::write2ConfigFiles(std::vector<std::ofstream*>& Files,
                                    char* pBuff, int FileIdx)
  {
  
      if (FileIdx == 0)
      {
          std::vector<std::ofstream*>::iterator aIt = Files.begin();
  
          while (aIt != Files.end())
          {
              std::ofstream* pCopy = (*aIt);
              pCopy->write(pBuff, strlen(pBuff));
              ++aIt;
          }
      }
      else
      {
          Files[FileIdx - 1]->write(pBuff, strlen(pBuff));
      }
  }
  
  //------------------------------------------------------------------------------
  
  void WECmdArgs::updateWithJobFile(int Idx)
  {
      setLocFile("");	// resetting the from the previous import
      std::string aJobFileName =  fVecJobFiles[Idx];
      setSchemaAndTableFromJobFile(aJobFileName);
      setEnclByAndEscCharFromJobFile(aJobFileName);
      setJobFileName(aJobFileName);
  
      std::ostringstream aSS;
      aSS << fOrigJobId << ".xml_" << (Idx + 1);
      fJobId = aSS.str();
  }
  
  
  //------------------------------------------------------------------------------
  
  std::string WECmdArgs::replaceCharInStr(const std::string& Str, char C, char R)
  {
      std::stringstream aSs;
  
      size_t start = 0, end = 0;
      end = Str.find_first_of(C);
  
      do
      {
          if (end != string::npos)
          {
              aSs << Str.substr(start, end - start) << R;
              start = end + 1;
          }
          else
          {
              aSs << Str.substr(start, end - start);
              break;
          }
  
          end = Str.find_first_of(C, start);
      }
      while (start != end);
  
      return aSs.str();
  }
  
  //------------------------------------------------------------------------------
  // Introduced to handle Bug 4342 with Mode 2
  
  std::string WECmdArgs::PrepMode2ListOfFiles(std::string& FileName)
  {
      VecArgs aInfileList;
      std::string bulkRootPath = getBulkRootDir();
      //cout << "Inside PrepMode2ListOfFiles("<< FileName << ")" << endl;
      std::string aFileName = FileName;
  
      istringstream iss(aFileName);
      size_t start = 0, end = 0;
      const char* sep = " ,|";
  
      end = aFileName.find_first_of(sep);
  
      do
      {
          if (end != string::npos)
          {
              std::string aFile = aFileName.substr(start, end - start);
  
              if (getDebugLvl() > 1)
                  cout << "File: " << aFileName.substr(start, end - start) << endl;
  
              start = end + 1;
              aInfileList.push_back(aFile);
          }
          else
          {
              std::string aFile = aFileName.substr(start, end - start);
  
              if (getDebugLvl() > 1)
                  cout << "Next Input File " << aFileName.substr(start, end - start) << endl;
  
              aInfileList.push_back(aFile);
              break;
          }
  
          end = aFileName.find_first_of(sep, start);
      }
      while (start != end);
  
      std::ostringstream aSS;
      int aVecSize = aInfileList.size();
      int aVecIdx = 0;
  
      // Take file list one by one and append it to one string
      while (aVecIdx < aVecSize)
      {
          std::string aNextFile = aInfileList[aVecIdx];
          aVecIdx++;
  
          //aInfileList.pop_front();
          if (aNextFile.at(0) != '/')
          {
              aSS << bulkRootPath << "/data/import/" + aNextFile;
          }
          else
          {
              aSS << aNextFile;
          }
  
          if (aVecIdx < aVecSize) aSS << ",";
      }
  
      //cout << "File list are = " << aSS.str() << endl;
  
      return aSS.str();
  }
  
  //------------------------------------------------------------------------------
  // Get set of column names in the "current" table being processed from the
  // Job xml file.
  //------------------------------------------------------------------------------
  void WECmdArgs::getColumnList( std::set<std::string>& columnList ) const
  {
      columnList.clear();
  
      for (unsigned k = 0; k < fColFldsFromJobFile.size(); k++)
      {
          columnList.insert( fColFldsFromJobFile[k] );
      }
  }
  
  //-----------------------------------------------------------------------------
  // check for the bulkload log directory. If it is not existing, create it
  // w.r.t Bug 6137
  //-----------------------------------------------------------------------------
  
  void WECmdArgs::checkForBulkLogDir(const std::string& BulkRoot)
  {
      if ( !boost::filesystem::exists(BulkRoot.c_str()) )
      {
          cout << "Creating directory : " << BulkRoot << endl;
          boost::filesystem::create_directories(BulkRoot.c_str());
      }
  
      if ( boost::filesystem::exists(BulkRoot.c_str()) )
      {
          // create the job directory also if not existing
          std::ostringstream aSS;
          aSS << BulkRoot;
          aSS << "/job";
          std::string jobDir = aSS.str();
  
          if ( !boost::filesystem::exists(jobDir.c_str()) )
          {
              cout << "Creating directory : " << jobDir << endl;
              bool aSuccess = boost::filesystem::create_directories(jobDir.c_str());
  
              if (!aSuccess)
              {
                  cout << "\nFailed to create job directory, check permissions\n" << endl;
                  throw runtime_error("Failed to create job directory, check permissions");
              }
          }
      }
      else
      {
          cout << "\nFailed to create bulk directory, check permissions\n" << endl;
          throw runtime_error("Failed to create bulk directory, check permissions");
      }
  }
  
  
  } /* namespace WriteEngine */
  