hdfs-scanner.h

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed 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.


#ifndef IMPALA_EXEC_HDFS_SCANNER_H_
#define IMPALA_EXEC_HDFS_SCANNER_H_

#include <vector>
#include <memory>
#include <stdint.h>
#include <boost/regex.hpp>
#include <boost/scoped_ptr.hpp>

#include "codegen/impala-ir.h"
#include "exec/hdfs-scan-node.h"
#include "exec/scan-node.h"
#include "exec/scanner-context.h"
#include "runtime/disk-io-mgr.h"
#include "runtime/row-batch.h"

namespace impala {

class Compression;
class DescriptorTbl;
class Expr;
class HdfsPartitionDescriptor;
class MemPool;
class SlotDescriptor;
class Status;
class TextConverter;
class Tuple;
class TupleDescriptor;
class TPlanNode;
class TScanRange;
class Codec;

struct FieldLocation {
  char* start;
  int len;

  static const char* LLVM_CLASS_NAME;
};

//
//
//
class HdfsScanner {
 public:
  const static int FILE_BLOCK_SIZE = 4096;

  HdfsScanner(HdfsScanNode* scan_node, RuntimeState* state);

  virtual ~HdfsScanner();

  virtual Status Prepare(ScannerContext* context);

  virtual Status ProcessSplit() = 0;

  virtual void Close();




  static const char* LLVM_CLASS_NAME;

 protected:
  HdfsScanNode* scan_node_;

  RuntimeState* state_;

  ScannerContext* context_;

  ScannerContext::Stream* stream_;

  std::vector<ExprContext*> conjunct_ctxs_;

  Tuple* template_tuple_;

  int tuple_byte_size_;

  Tuple* tuple_;

  RowBatch* batch_;

  uint8_t* tuple_mem_;

  int num_errors_in_file_;

  boost::scoped_ptr<TextConverter> text_converter_;

  int32_t num_null_bytes_;

  Status parse_status_;

  boost::scoped_ptr<Codec> decompressor_;

  THdfsCompression::type decompression_type_;

  boost::scoped_ptr<MemPool> data_buffer_pool_;

  RuntimeProfile::Counter* decompress_timer_;

  typedef int (*WriteTuplesFn)(HdfsScanner*, MemPool*, TupleRow*, int, FieldLocation*,
      int, int, int, int);
  WriteTuplesFn write_tuples_fn_;

  Status InitializeWriteTuplesFn(HdfsPartitionDescriptor* partition,
      THdfsFileFormat::type type, const std::string& scanner_name);

  void StartNewRowBatch();

  virtual Status InitNewRange() = 0;

  int GetMemory(MemPool** pool, Tuple** tuple_mem, TupleRow** tuple_row_mem);

  Status CommitRows(int num_rows);

  void AddFinalRowBatch();

  void AttachPool(MemPool* pool, bool commit_batch) {
    DCHECK(batch_ != NULL);
    DCHECK(pool != NULL);
    batch_->tuple_data_pool()->AcquireData(pool, false);
    if (commit_batch) CommitRows(0);
  }

  bool IR_ALWAYS_INLINE EvalConjuncts(TupleRow* row)  {
    return ExecNode::EvalConjuncts(&conjunct_ctxs_[0], conjunct_ctxs_.size(), row);
  }

  int WriteEmptyTuples(RowBatch* row_batch, int num_tuples);

  int WriteEmptyTuples(ScannerContext* context, TupleRow* tuple_row, int num_tuples);

  int WriteAlignedTuples(MemPool* pool, TupleRow* tuple_row_mem, int row_size,
      FieldLocation* fields, int num_tuples,
      int max_added_tuples, int slots_per_tuple, int row_start_indx);

  Status UpdateDecompressor(const THdfsCompression::type& compression);
  Status UpdateDecompressor(const std::string& codec);

  bool ReportTupleParseError(FieldLocation* fields, uint8_t* errors, int row_idx);

  virtual void LogRowParseError(int row_idx, std::stringstream*);

  //
  //
  bool WriteCompleteTuple(MemPool* pool, FieldLocation* fields, Tuple* tuple,
      TupleRow* tuple_row, Tuple* template_tuple, uint8_t* error_fields,
      uint8_t* error_in_row);

  static llvm::Function* CodegenWriteCompleteTuple(HdfsScanNode*, LlvmCodeGen*,
      const std::vector<ExprContext*>& conjunct_ctxs);

  static llvm::Function* CodegenWriteAlignedTuples(HdfsScanNode*, LlvmCodeGen*,
      llvm::Function* write_tuple_fn);

  void ReportColumnParseError(const SlotDescriptor* desc, const char* data, int len);

  void InitTuple(Tuple* template_tuple, Tuple* tuple) {
    if (template_tuple != NULL) {
      memcpy(tuple, template_tuple, tuple_byte_size_);
    } else {
      memset(tuple, 0, sizeof(uint8_t) * num_null_bytes_);
    }
  }

  inline Tuple* next_tuple(Tuple* t) const {
    uint8_t* mem = reinterpret_cast<uint8_t*>(t);
    return reinterpret_cast<Tuple*>(mem + tuple_byte_size_);
  }

  inline TupleRow* next_row(TupleRow* r) const {
    uint8_t* mem = reinterpret_cast<uint8_t*>(r);
    return reinterpret_cast<TupleRow*>(mem + batch_->row_byte_size());
  }

  ExprContext* GetConjunctCtx(int idx) const;
};

}

#endif