// **********************************************************************
// @@@ 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 @@@
// **********************************************************************
#ifndef COM_HDFS_SCAN_H
#define COM_HDFS_SCAN_H

#include "ComTdb.h"
#include "ExpLOBinterface.h"
#include "ComQueue.h"

//
// Task Definition Block
//
class ComTdbHdfsScan : public ComTdb
{
  friend class ExHdfsScanTcb;
  friend class ExHdfsScanPrivateState;
  friend class ExOrcScanTcb;
  friend class ExOrcScanPrivateState;
  friend class ExOrcFastAggrTcb;
  friend class ExOrcFastAggrPrivateState;

 protected:
  enum
  {
    USE_CURSOR_MULTI            = 0x0001,
    DO_SPLIT_FILE_OPT           = 0x0002,
    HDFS_PREFETCH               = 0x0004,
    // flag to indicate whther we need to use Compressed Internal Format
    USE_CIF                     = 0x0008,
    // flag to indicate whther we need to use Defragmentation  with Compressed Internal Format
    USE_CIF_DEFRAG              = 0x0010,

    // ignore conversion errors and continue reading the next row.
    CONTINUE_ON_ERROR           = 0x0020,
    LOG_ERROR_ROWS              = 0x0040,
    ASSIGN_RANGES_AT_RUNTIME    = 0x0080,
    TREAT_EMPTY_AS_NULL         = 0x0100,
    USE_LIBHDFS_SCAN            = 0x0200,
    COMPRESSED_FILE             = 0x0400
  };

  // Expression to filter rows.
  ExExprPtr selectPred_;                                     // 00 - 07

  // Expression to move selected rows into buffer pool.
  ExExprPtr moveExpr_;                                       // 08 - 15

  // Convert selection pred cols in HDFS row from exploded format 
  // with all types in ascii to aligned format with the correct 
  // binary type for each column 
  ExExprPtr convertExpr_;                                    // 16 - 23

  // Convert move expr only cols in HDFS row from exploded format 
  // with all types in ascii to aligned format with the correct 
  // binary type for each column 
  ExExprPtr moveColsConvertExpr_;                            // 24 - 31

  NABasicPtr hostName_;                                      // 32 - 39

  ExCriDescPtr workCriDesc_;                                 // 40 - 47

  Int64 moveExprColsRowLength_;                              // 48 - 55
  UInt16 moveExprColsTuppIndex_;                             // 56 - 57

  short type_;                                      
  char filler1_[4];                                          // 58 - 63

  // I/O buffer for interface to hdfs access layer.
  Int64 hdfsBufSize_;                                        // 64 - 71

  // max length of the row converted from hdfs to SQL format, before 
  // projection.
  Int64 hdfsSqlMaxRecLen_;                                   // 72 - 79

  // max length of output row
  Int64 outputRowLength_;                                    // 80 - 87

  // index into atp of output row tupp
  UInt16 tuppIndex_;                                         // 88 - 89

  UInt16 workAtpIndex_;                                      // 90 - 91

  char recordDelimiter_;                                     // 92 - 92

  char columnDelimiter_;                                     // 93 - 93

  // index into atp of source ascii row(hive, hdfs) which will be converted to sql row
  UInt16 asciiTuppIndex_;                                    // 94 - 95

  UInt32 flags_;                                             // 96 - 99

  UInt16 port_;                                              // 100 - 101

  UInt16 convertSkipListSize_;                               // 102 - 103
  Int16Ptr convertSkipList_;                                 // 104 - 111

  Int64 asciiRowLen_;                                        // 112 - 119

  // cumulative list of all scanInfo entries for all esps (nodeMap entries)
  QueuePtr hdfsFileInfoList_;                                // 120-127

  // range of files in hdfsFileInfoList_ that each esp need to read from.
  // start at 'Begin' entry and read 'Num' entries.
  // At runtime, each esp will pick one of the ranges.
  // Later, we may align esps with the local files.
  QueuePtr hdfsFileRangeBeginList_;                          // 128 - 135
  QueuePtr hdfsFileRangeNumList_;                            // 136 - 143

  NABasicPtr tableName_;                                     // 144 - 151
  UInt32 rangeTailIOSize_;                                   // 152 - 155
  UInt32 maxErrorRows_;                                      // 156 - 159
  NABasicPtr errCountTable_;                                  // 160 - 167
  NABasicPtr loggingLocation_;                                // 168 - 175
  NABasicPtr errCountRowId_;                                  // 176 - 183
  UInt32  hiveScanMode_;                                      // 184 - 187
  UInt16 origTuppIndex_;                                      // 188 - 189
  char fillersComTdbHdfsScan1_[2];                            // 190 - 191
  NABasicPtr nullFormat_;                                     // 192 - 199

  // next 4 params are used to check if data under hdfsFileDir
  // was modified after query was compiled.
  NABasicPtr hdfsRootDir_;                                     // 200 - 207
  Int64  modTSforDir_;                                         // 208 - 215
  Lng32  numOfPartCols_;                                       // 216 - 219
  Lng32  hdfsIoByteArraySizeInKB_;                             // 220 - 223
  QueuePtr hdfsDirsToCheck_;                                   // 224 - 231
    
public:
  enum HDFSFileType
  {
    UNKNOWN_ = 0,
    TEXT_ = 1,
    SEQUENCE_ = 2,
    ORC_ = 3
  };

  // Constructor
  ComTdbHdfsScan(); // dummy constructor. Used by 'unpack' routines.
  
  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 * hdfsHostName,
                 tPort  hdfsPort, 
		 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 = NULL,
                 char * loggingLocation = NULL,
                 char * errCountId = NULL,

                 // next 4 params are used to check if data under hdfsFileDir
                 // was modified after query was compiled.
                 char * hdfsRootDir  = NULL,
                 Int64  modTSforDir   = -1,
                 Lng32  numOfPartCols = -1,
                 Queue * hdfsDirsToCheck = NULL
                 );

  ~ComTdbHdfsScan();

  //----------------------------------------------------------------------
  // Redefine virtual functions required for Versioning.
  //----------------------------------------------------------------------
  virtual unsigned char getClassVersionID()
  {
    return 1;
  }

  virtual void populateImageVersionIDArray()
  {
    setImageVersionID(1,getClassVersionID());
    ComTdb::populateImageVersionIDArray();
  }

  virtual short getClassSize() { return (short)sizeof(ComTdbHdfsScan); }  
  Long pack(void *);

  Lng32 unpack(void *, void * reallocator);
  
  void display() const;

  inline ComTdb * getChildTdb();

  Int32 orderedQueueProtocol() const;

  char * tableName() { return tableName_; }
  char * getErrCountTable() { return errCountTable_; }
  void   setErrCountTable(char * v ) { errCountTable_ = v; }
  char * getLoggingLocation() { return loggingLocation_; }
  void   setLoggingLocation(char * v ) { loggingLocation_ = v; }
  char * getErrCountRowId() { return errCountRowId_; }
  void   setErrCountRowId(char * v ) { errCountRowId_ = v; }
  void   setHiveScanMode(UInt32 v ) { hiveScanMode_ = v; }
  UInt32 getHiveScanMode() { return hiveScanMode_; }

  Queue* getHdfsFileInfoList() const {return hdfsFileInfoList_;}
  Queue* getHdfsFileRangeBeginList() {return hdfsFileRangeBeginList_;}
  Queue* getHdfsFileRangeNumList() {return hdfsFileRangeNumList_;}

  char * getNullFormat() { return nullFormat_; }

  const NABoolean isTextFile() const { return (type_ == TEXT_);}
  const NABoolean isSequenceFile() const { return (type_ == SEQUENCE_);}  
  const NABoolean isOrcFile() const { return (type_ == ORC_);}

  void setUseCursorMulti(NABoolean v)
  {(v ? flags_ |= USE_CURSOR_MULTI : flags_ &= ~USE_CURSOR_MULTI); };
  NABoolean useCursorMulti() { return (flags_ & USE_CURSOR_MULTI) != 0; };

  void setDoSplitFileOpt(NABoolean v)
  {(v ? flags_ |= DO_SPLIT_FILE_OPT : flags_ &= ~DO_SPLIT_FILE_OPT); };
  NABoolean doSplitFileOpt() { return (flags_ & DO_SPLIT_FILE_OPT) != 0; };

  void setHdfsPrefetch(NABoolean v)
  {(v ? flags_ |= HDFS_PREFETCH : flags_ &= ~HDFS_PREFETCH); };
  NABoolean hdfsPrefetch() { return (flags_ & HDFS_PREFETCH) != 0; };

  void setUseCif(NABoolean v)
  {(v ? flags_ |= USE_CIF : flags_ &= ~USE_CIF); };
  NABoolean useCif() { return (flags_ & USE_CIF) != 0; };
  
  void setUseCifDefrag(NABoolean v)
  {(v ? flags_ |= USE_CIF_DEFRAG : flags_ &= ~USE_CIF_DEFRAG); };
  NABoolean useCifDefrag() { return (flags_ & USE_CIF_DEFRAG) != 0; };

  void setContinueOnError(NABoolean v)
  {(v ? flags_ |= CONTINUE_ON_ERROR : flags_ &= ~CONTINUE_ON_ERROR); };
  NABoolean continueOnError() { return (flags_ & CONTINUE_ON_ERROR) != 0; };
  
  void setLogErrorRows(NABoolean v)
  {(v ? flags_ |= LOG_ERROR_ROWS : flags_ &= ~LOG_ERROR_ROWS); };
  NABoolean getLogErrorRows() { return (flags_ & LOG_ERROR_ROWS) != 0; };
  
  void setAssignRangesAtRuntime(NABoolean v)
  {(v ? flags_ |= ASSIGN_RANGES_AT_RUNTIME : flags_ &= ~ASSIGN_RANGES_AT_RUNTIME); }
  NABoolean getAssignRangesAtRuntime() const
                                { return (flags_ & ASSIGN_RANGES_AT_RUNTIME) != 0; }

  void setUseLibhdfsScan(NABoolean v)
  {(v ? flags_ |= USE_LIBHDFS_SCAN : flags_ &= ~USE_LIBHDFS_SCAN); }
  NABoolean getUseLibhdfsScan() const
                                { return (flags_ & USE_LIBHDFS_SCAN) != 0; }

  void setCompressedFile(NABoolean v)
  {(v ? flags_ |= COMPRESSED_FILE : flags_ &= ~COMPRESSED_FILE); }
  NABoolean isCompressedFile() const
                                { return (flags_ & COMPRESSED_FILE) != 0; }

  UInt32 getMaxErrorRows() const { return maxErrorRows_;}
  void setMaxErrorRows(UInt32 v ) { maxErrorRows_= v; }
  
  // ---------------------------------------------------------------------
  // Used by the internal SHOWPLAN command to get attributes of a TDB.
  // ---------------------------------------------------------------------
  virtual void displayContents(Space *space,ULng32 flag);

  virtual const ComTdb* getChild(Int32 pos) const;

  virtual Int32 numChildren() const { return 0; }

  virtual const char *getNodeName() const { return "EX_HDFS_SCAN"; };

  virtual Int32 numExpressions() const { return 4; }

  virtual ex_expr* getExpressionNode(Int32 pos) {
     if (pos == 0)
	return selectPred_;
     else if (pos == 1)
	return moveExpr_;
     else if (pos == 2)
       return convertExpr_;
     else if (pos == 3)
       return moveColsConvertExpr_;
     else
       return NULL;
  }
  
  virtual const char * getExpressionName(Int32 pos) const {
     if (pos == 0)
	return "selectExpr_";
     else if (pos == 1)
	return "moveExpr_";
     else if (pos == 2)
       return "convertExpr_";
     else if (pos ==3)
       return "moveColsConvertExpr_";

     else
	return NULL;
  }
  
  ExpTupleDesc *getHdfsSqlRowDesc() const
  {
    return workCriDesc_->getTupleDescriptor(workAtpIndex_);
  }

  ExpTupleDesc *getHdfsAsciiRowDesc() const
  {
    return workCriDesc_->getTupleDescriptor(asciiTuppIndex_);
  }

  ExpTupleDesc *getHdfsOrigRowDesc() const
  {
    return workCriDesc_->getTupleDescriptor(origTuppIndex_);
  }

  ExpTupleDesc *getMoveExprColsRowDesc() const
  {
    return workCriDesc_->getTupleDescriptor(moveExprColsTuppIndex_);
  }

  Queue * hdfsDirsToCheck() { return hdfsDirsToCheck_; }
 
  char *hdfsRootDir() { return hdfsRootDir_; }
  void setHdfsIoByteArraySize(int size)
    { hdfsIoByteArraySizeInKB_ = size; }
  int getHdfsIoByteArraySize() 
    { return hdfsIoByteArraySizeInKB_; }
};

inline ComTdb * ComTdbHdfsScan::getChildTdb()
{
  return NULL;
};

/*****************************************************************************
  Description : Return ComTdb* depending on the position argument.
                  Position 0 means the left most child.
  Comments    :
*****************************************************************************/
inline const ComTdb* ComTdbHdfsScan::getChild(Int32 pos) const
{
  return NULL;
};

// ComTdbOrcFastAggr 
class ComTdbOrcFastAggr : public ComTdbHdfsScan
{
  friend class ExOrcFastAggrTcb;
  friend class ExOrcFastAggrPrivateState;

public:
  enum OrcAggrType
  {
    UNKNOWN_ = 0,
    COUNT_      = 1,
    MIN_           = 2,
    MAX_          = 3
  };

  // Constructor
  ComTdbOrcFastAggr(); // dummy constructor. Used by 'unpack' routines.
  
  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
                );
  
  ~ComTdbOrcFastAggr();

  virtual short getClassSize() { return (short)sizeof(ComTdbOrcFastAggr); }  

  virtual const char *getNodeName() const { return "EX_ORC_FAST_AGGR"; };

 private:
   OrcAggrType type_;
};

#endif