// **********************************************************************
// @@@ START COPYRIGHT @@@
//
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
//
// @@@ END COPYRIGHT @@@
// **********************************************************************

#include "ComTdbHdfsScan.h"
#include "ComTdbCommon.h"

// Dummy constructor for "unpack" routines.
ComTdbHdfsScan::ComTdbHdfsScan():
 ComTdb(ComTdb::ex_HDFS_SCAN, eye_HDFS_SCAN)
{};

// Constructor

ComTdbHdfsScan::ComTdbHdfsScan(
                               char * tableName,
                               short type,
                               ex_expr * select_pred,
                               ex_expr * move_expr,
                               ex_expr * convert_expr,
                               ex_expr * move_convert_expr,
                               UInt16 convertSkipListSize,
                               Int16 * convertSkipList,
                               char * hostName,
                               tPort port,
                               Queue * hdfsFileInfoList,
                               Queue * hdfsFileRangeBeginList,
                               Queue * hdfsFileRangeNumList,
                               char recordDelimiter,
                               char columnDelimiter,
                               char * nullFormat,
                               Int64 hdfsBufSize,
                               UInt32 rangeTailIOSize,
                               Int64 hdfsSqlMaxRecLen,
                               Int64 outputRowLength,
                               Int64 asciiRowLen,
                               Int64 moveColsRowLen,
                               const unsigned short tuppIndex,
                               const unsigned short asciiTuppIndex,
                               const unsigned short workAtpIndex,
                               const unsigned short moveColsTuppIndex,
                               const unsigned short origTuppIndex,
                               ex_cri_desc * work_cri_desc,
                               ex_cri_desc * given_cri_desc,
                               ex_cri_desc * returned_cri_desc,
                               queue_index down,
                               queue_index up,
                               Cardinality estimatedRowCount,
                               Int32  numBuffers,
                               UInt32  bufferSize,
                               char * errCountTable,
                               char * loggingLocation,
                               char * errCountId,

                               char * hdfsRootDir,
                               Int64  modTSforDir,
                               Lng32  numOfPartCols,
                               Queue * hdfsDirsToCheck

                               )
: ComTdb( ComTdb::ex_HDFS_SCAN,
            eye_HDFS_SCAN,
            estimatedRowCount,
            given_cri_desc,
            returned_cri_desc,
            down,
            up,
            numBuffers,        // num_buffers - we use numInnerTuples_ instead.
            bufferSize),       // buffer_size - we use numInnerTuples_ instead.
  tableName_(tableName),
  type_(type),
  selectPred_(select_pred),
  moveExpr_(move_expr),
  convertExpr_(convert_expr),
  moveColsConvertExpr_(move_convert_expr),
  convertSkipListSize_(convertSkipListSize),
  convertSkipList_(convertSkipList),
  hostName_(hostName),
  port_(port),
  hdfsFileInfoList_(hdfsFileInfoList),
  hdfsFileRangeBeginList_(hdfsFileRangeBeginList),
  hdfsFileRangeNumList_(hdfsFileRangeNumList),
  recordDelimiter_(recordDelimiter),
  columnDelimiter_(columnDelimiter),
  nullFormat_(nullFormat),
  hdfsBufSize_(hdfsBufSize),
  rangeTailIOSize_(rangeTailIOSize),
  hdfsSqlMaxRecLen_(hdfsSqlMaxRecLen),
  outputRowLength_(outputRowLength),
  asciiRowLen_(asciiRowLen),
  moveExprColsRowLength_(moveColsRowLen),
  origTuppIndex_(origTuppIndex),
  tuppIndex_(tuppIndex),
  asciiTuppIndex_(asciiTuppIndex),
  workAtpIndex_(workAtpIndex),
  moveExprColsTuppIndex_(moveColsTuppIndex),
  workCriDesc_(work_cri_desc),
  flags_(0),
  errCountTable_(errCountTable),
  loggingLocation_(loggingLocation),
  errCountRowId_(errCountId),
  hdfsRootDir_(hdfsRootDir),
  modTSforDir_(modTSforDir),
  numOfPartCols_(numOfPartCols),
  hdfsDirsToCheck_(hdfsDirsToCheck)
{};

ComTdbHdfsScan::~ComTdbHdfsScan()
{};

void ComTdbHdfsScan::display() const 
{};

Long ComTdbHdfsScan::pack(void * space)
{
  tableName_.pack(space);
  selectPred_.pack(space);
  moveExpr_.pack(space);
  convertExpr_.pack(space);
  moveColsConvertExpr_.pack(space);
  convertSkipList_.pack(space);
  hostName_.pack(space);
  workCriDesc_.pack(space);

  // pack elements in hdfsFileInfoList
  getHdfsFileInfoList()->position();
  for (Lng32 i = 0; i < getHdfsFileInfoList()->numEntries(); i++)
    {
      HdfsFileInfo * hdf = (HdfsFileInfo*)getHdfsFileInfoList()->getNext();
      
      hdf->fileName_.pack(space);
    }

  hdfsFileInfoList_.pack(space);
  hdfsFileRangeBeginList_.pack(space);
  hdfsFileRangeNumList_.pack(space);

  nullFormat_.pack(space);

  errCountTable_.pack(space);
  loggingLocation_.pack(space);
  errCountRowId_.pack(space);

  hdfsRootDir_.pack(space);
  hdfsDirsToCheck_.pack(space);

  return ComTdb::pack(space);
}

Lng32 ComTdbHdfsScan::unpack(void * base, void * reallocator)
{
  if (tableName_.unpack(base)) return -1;
  if(selectPred_.unpack(base, reallocator)) return -1;
  if(moveExpr_.unpack(base, reallocator)) return -1;
  if(convertExpr_.unpack(base, reallocator)) return -1;
  if(moveColsConvertExpr_.unpack(base, reallocator)) return -1;
  if(convertSkipList_.unpack(base)) return -1;
  if(hostName_.unpack(base)) return -1;
  if(workCriDesc_.unpack(base, reallocator)) return -1;

  if (hdfsFileInfoList_.unpack(base, reallocator)) return -1;

  // unpack elements in hdfsFileInfoList
  getHdfsFileInfoList()->position();
  for (Lng32 i = 0; i < getHdfsFileInfoList()->numEntries(); i++)
    {
      HdfsFileInfo * hdf = (HdfsFileInfo*)getHdfsFileInfoList()->getNext();
      
      if (hdf->fileName_.unpack(base)) return -1;
    }

  if (hdfsFileRangeBeginList_.unpack(base, reallocator)) return -1;
  if (hdfsFileRangeNumList_.unpack(base, reallocator)) return -1;

  if (nullFormat_.unpack(base)) return -1;

  if (errCountTable_.unpack(base)) return -1;
  if (loggingLocation_.unpack(base)) return -1;
  if (errCountRowId_.unpack(base)) return -1;

  if (hdfsRootDir_.unpack(base)) return -1;
  if (hdfsDirsToCheck_.unpack(base, reallocator)) return -1;

  return ComTdb::unpack(base, reallocator);
}

void ComTdbHdfsScan::displayContents(Space * space,ULng32 flag)
{
  ComTdb::displayContents(space,flag & 0xFFFFFFFE);

  if(flag & 0x00000008)
    {
      char buf[16384];

      str_sprintf(buf, "\nFor ComTdbHdfsScan :");
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      str_sprintf(buf, "tableName_ = %s", (char*)tableName_);
     space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      if (hostName_)
        str_sprintf(buf, "hostName_ = %s, port_ = %d", 
                       (char *) hostName_, port_);
      else
        str_sprintf(buf, "hostName_ = NULL (empty scan)");
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      str_sprintf(buf, "recordDelimiter_ = %d, columnDelimiter_ = %d", 
                         recordDelimiter_ ,  columnDelimiter_ );
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      str_sprintf(buf, "hdfsBufSize_ = %d, rangeTailIOSize_ = %d, "
                       "hdfsSqlMaxRecLen_ = %d",
                        hdfsBufSize_ ,  rangeTailIOSize_, hdfsSqlMaxRecLen_ );
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      str_sprintf(buf, "tuppIndex_ = %d, workAtpIndex_ = %d",
                          tuppIndex_ ,      workAtpIndex_ );
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      str_sprintf(buf, "asciiTuppIndex_ = %d, asciiRowLen_ = %d",
                          asciiTuppIndex_ ,      asciiRowLen_);
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      str_sprintf(buf, "moveExprColsTuppIndex_ = %d, moveExprColsRowLength_ = %d",
                          moveExprColsTuppIndex_ ,      moveExprColsRowLength_);
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      if (convertSkipListSize_)
        {
          // only the first element of the skip list is printed for now
          str_sprintf(buf, "convertSkipListSize_ = %d, convertSkipList_ = %d",
                      convertSkipListSize_ ,      convertSkipList_[0]);
          space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));
        }

      str_sprintf(buf, "outputRowLength_ = %d", outputRowLength_);
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      str_sprintf(buf, "Flag = %b",flags_);
      space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

      Queue *hdfsFileInfoList = hdfsFileInfoList_;
      if (hdfsFileInfoList)
        {
          UInt32 dataElems = hdfsFileInfoList->numEntries();
          str_sprintf(buf, "\nNumber of ranges to scan: %d",
                      (Int32) dataElems);
          space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

          Lng32 currInstNum = -1;
          Lng32 currEntryNum = 0;
          Lng32 startOfNextInst = 0;
          HdfsFileInfo * hdfo;

          hdfsFileInfoList->position();
          while ((hdfo = (HdfsFileInfo*)hdfsFileInfoList->getNext()) != NULL)
            {
              if (currEntryNum >= startOfNextInst)
                {
                  currInstNum++;
                  startOfNextInst = 
                    *((Lng32 *) getHdfsFileRangeBeginList()->get(currInstNum)) +
                    *((Lng32 *) getHdfsFileRangeNumList()->get(currInstNum));
                }
	      currEntryNum++;
            }

         Int32 numESPs = currInstNum+1;
         str_sprintf(buf, "Number of esps to scan:    %d\n",
                      numESPs);
          space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

	  currInstNum = -1;
          currEntryNum = 0;
          startOfNextInst = 0;

          Int64 bytesPerESP[4096];
          Int64 totalBytes = 0;

          for (Int32 i=0; i<4096; i++) 
	    bytesPerESP[i] = 0;

          if (isOrcFile())
            str_sprintf(
                        buf, 
                        "%6s %6s %12s %12s  %s",
                        "  Esp#", "Range#", " StartRowNum", "     NumRows", "    FileName");
          else
            str_sprintf(
                        buf, 
                        "%6s %6s %12s %12s  %s",
                        "  Esp#", "Range#", " StartOffset", "   BytesRead", "    FileName");

           space->allocateAndCopyToAlignedSpace(buf, str_len(buf),
                                                   sizeof(short));
   
	  str_sprintf(
		      buf, "====== ====== ============ ============  ==============================\n");
           space->allocateAndCopyToAlignedSpace(buf, str_len(buf),
                                                   sizeof(short));

          hdfsFileInfoList->position();
          while ((hdfo = (HdfsFileInfo*)hdfsFileInfoList->getNext()) != NULL)
            {
              if (currEntryNum >= startOfNextInst)
                {
                  currInstNum++;
                  startOfNextInst = 
                    *((Lng32 *) getHdfsFileRangeBeginList()->get(currInstNum)) +
                    *((Lng32 *) getHdfsFileRangeNumList()->get(currInstNum));
                }

	      NABoolean printStr = FALSE;
	      NABoolean isLocal = FALSE;
	      char splitInfo[200];
	      strcpy(splitInfo, " ");
              if (hdfo->fileIsLocal()) //hdfo->getFlags() & HdfsFileInfo::HDFSFILEFLAGS_LOCAL)
		{
		  isLocal = TRUE;
		  strcpy(splitInfo, "(local");
		}

	      if ((hdfo->fileIsSplitBegin()) ||
		  (hdfo->fileIsSplitEnd()))
		{
		  if (isLocal)
		    strcat(splitInfo, ", ");
		  else
		    strcpy(splitInfo, "(");

		  printStr = TRUE;
		  strcat(splitInfo, "split_");
		  if (hdfo->fileIsSplitBegin())
		    {
		      strcat(splitInfo, "b");
		    
		      if (hdfo->fileIsSplitEnd())
			strcat(splitInfo, "/");
		    }

		  if (hdfo->fileIsSplitEnd())
		    strcat(splitInfo, "e");

		  strcat(splitInfo, ")");
		}
	      else if (isLocal)
		strcat(splitInfo, ")");

              if (isSequenceFile())
                {
                  if ( isLocal ||
                       (hdfo->fileIsSplitBegin()) ||
                       (hdfo->fileIsSplitEnd()))
                     strcat(splitInfo, ", ");
                     
                  strcat(splitInfo, "seq");
                }
                
	      // filename is fully qualified addr and looks like this:
	      // hdfs://....com/hive/warehouse/so_dtl_f_test/000030_0
	      // To make it little easier to read, if it starts with "hdfs://", then pick
	      // the last 2 parts, or maybe 3 parts at some point.
	      Lng32 i = 0;
	      if (str_cmp(hdfo->fileName(), "hdfs://", strlen("hdfs://")) == 0)
		{
		  i = strlen(hdfo->fileName()) - 1;
		  Lng32 numParts = 2;
		  NABoolean done = FALSE;
		  while (NOT done)
		    {
		      if ((i > 0) && (numParts > 0) && (hdfo->fileName()[i] != '/'))
			i--;
		      else
			{
			  if (i <= 0)
			    done = TRUE;
			  else 
			    {
			      numParts--;
			      if (numParts == 0)
				{
				  i++;
				  done = TRUE;
				}
			      else
				i--;
			    } // else
			} // else
		    } // while
		} // if

              str_sprintf(
                   buf, 
		   "%6d %6d %12Ld %12Ld  %s %s",
                   currInstNum,
                   currEntryNum,
                   hdfo->getStartOffset(),
                   hdfo->getBytesToRead(),
                   &hdfo->fileName()[i],
		   splitInfo);

              space->allocateAndCopyToAlignedSpace(buf, str_len(buf),
                                                   sizeof(short));
              bytesPerESP[currInstNum] += (hdfo->getBytesToRead() >= 0 ? hdfo->getBytesToRead() : 0);
              totalBytes += (hdfo->getBytesToRead() >= 0 ? hdfo->getBytesToRead() : 0);
              currEntryNum++;
            }

          str_sprintf(buf, "\nSummary of bytes read per ESP (%Ld = 100 percent):\n",
                      totalBytes);
          space->allocateAndCopyToAlignedSpace(buf, str_len(buf),
                                               sizeof(short));

          if (totalBytes == 0)
            totalBytes = 1; // avoid divide by zero below

          numESPs = currInstNum+1;

          for (Int32 e=0; e<numESPs; e++)
            {
              str_sprintf(buf,"ESP %4d reads %18Ld bytes (%4d percent of avg)",
                          e,
                          bytesPerESP[e],
                          (Int32) 100 * bytesPerESP[e] * numESPs / totalBytes);
              space->allocateAndCopyToAlignedSpace(buf, str_len(buf),
                                                   sizeof(short));
            }
        }

      if (hdfsRootDir_)
        {
          str_sprintf(buf, "hdfsRootDir: %s", hdfsRootDir_);
          space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

          str_sprintf(buf, "modTSforDir_ = %Ld, numOfPartCols_ = %d",
                      modTSforDir_, numOfPartCols_);
          space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));

          if (hdfsDirsToCheck())
            {
              hdfsDirsToCheck()->position();
              char * dir = NULL;
              while ((dir = (char*)hdfsDirsToCheck()->getNext()) != NULL)
                {
                  str_sprintf(buf, "Dir Name: %s", dir);
                  space->allocateAndCopyToAlignedSpace(buf, str_len(buf), sizeof(short));
                }
            }

        }

    }

  if(flag & 0x00000001)
    {
      displayExpression(space,flag);
      displayChildren(space,flag);
    }
}

Int32 ComTdbHdfsScan::orderedQueueProtocol() const
{
  return 1;
}

///////////////////////////////////////////////////////////////
// ComTdbOrcAggr
///////////////////////////////////////////////////////////////

// Dummy constructor for "unpack" routines.
ComTdbOrcFastAggr::ComTdbOrcFastAggr():
 ComTdbHdfsScan()
{
  setNodeType(ComTdb::ex_ORC_AGGR);
  setEyeCatcher(eye_ORC_AGGR);
};

// Constructor
ComTdbOrcFastAggr::ComTdbOrcFastAggr(
                                     char * tableName,
                                     OrcAggrType type,
                                     Queue * hdfsFileInfoList,
                                     Queue * hdfsFileRangeBeginList,
                                     Queue * hdfsFileRangeNumList,
                                     ex_expr * proj_expr,
                                     Int64 projRowLen,
                                     const unsigned short projTuppIndex,
                                     const unsigned short returnedTuppIndex,
                                     ex_cri_desc * work_cri_desc,
                                     ex_cri_desc * given_cri_desc,
                                     ex_cri_desc * returned_cri_desc,
                                     queue_index down,
                                     queue_index up,
                                     Int32  numBuffers,
                                     UInt32  bufferSize
                                     )
  : ComTdbHdfsScan( 
                   tableName,
                   type,
                   NULL, NULL,
                   proj_expr,
                   NULL,
                   0, NULL, NULL, 0,
                   hdfsFileInfoList,
                   hdfsFileRangeBeginList,
                   hdfsFileRangeNumList,
                   0, 0, NULL, 0, 0, 0,
                   projRowLen, 
                   0, 0,
                   returnedTuppIndex,
                   0, 
                   projTuppIndex, 
                   0, 0,
                   work_cri_desc,
                   given_cri_desc,
                   returned_cri_desc,
                   down,
                   up,
                   0,
                   numBuffers,        // num_buffers - we use numInnerTuples_ instead.
                   bufferSize),       // buffer_size - we use numInnerTuples_ instead.
    type_(type)
{
  setNodeType(ComTdb::ex_ORC_AGGR);
  setEyeCatcher(eye_ORC_AGGR);
}

ComTdbOrcFastAggr::~ComTdbOrcFastAggr()
{
}