hdfs-sequence-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_SEQUENCE_SCANNER_H
#define IMPALA_EXEC_HDFS_SEQUENCE_SCANNER_H

//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//

#include "exec/base-sequence-scanner.h"

namespace impala {

class DelimitedTextParser;

class HdfsSequenceScanner : public BaseSequenceScanner {
 public:
  static const uint8_t SEQFILE_VERSION_HEADER[4];

  HdfsSequenceScanner(HdfsScanNode* scan_node, RuntimeState* state);

  virtual ~HdfsSequenceScanner();
  
  virtual Status Prepare(ScannerContext* context);

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

 protected:
  virtual FileHeader* AllocateFileHeader();
  virtual Status ReadFileHeader();
  virtual Status InitNewRange();
  virtual Status ProcessRange();
  
  virtual THdfsFileFormat::type file_format() const { 
    return THdfsFileFormat::SEQUENCE_FILE; 
  }

 private:
  const static int MAX_BLOCK_SIZE = (1024 * 1024 * 1024);

  static const char* const SEQFILE_VALUE_CLASS_NAME;

  Status ReadBlockHeader();

  Status ProcessBlockCompressedScanRange();

  Status ReadCompressedBlock();

  Status ProcessDecompressedBlock();

  Status GetRecord(uint8_t** record_ptr, int64_t *record_len);
  
  virtual void LogRowParseError(int row_idx, std::stringstream*);
  
  boost::scoped_ptr<DelimitedTextParser> delimited_text_parser_;
  std::vector<FieldLocation> field_locations_;

  struct SeqFileHeader : public BaseSequenceScanner::FileHeader {
    bool is_row_compressed;
  };

  struct RecordLocation {
    uint8_t* record;
    int64_t len;
  };

  std::vector<RecordLocation> record_locations_;

  int current_block_length_;

  int current_key_length_;

  uint8_t* unparsed_data_buffer_;

  int64_t num_buffered_records_in_compressed_block_;

  uint8_t* next_record_in_compressed_block_;
};

} // namespace impala

#endif // IMPALA_EXEC_HDFS_SEQUENCE_SCANNER_H