hdfs-lzo-text-scanner.h

// Copyright (c) 2012 Cloudera, Inc. All rights reserved.

#ifndef IMPALA_LZO_TEXT_SCANNER_H
#define IMPALA_LZO_TEXT_SCANNER_H

#include "lzo-header.h"
#include <boost/thread/locks.hpp>
#include "common/version.h"
#include "exec/hdfs-text-scanner.h"
#include "runtime/string-buffer.h"

// This provides support for reading files compressed with lzop.
// The file consists of a header and compressed blocks preceeded
// by their compressed and uncompressed block sizes.
//
// The following is a pseudo-BNF grammar for LZOFile. Comments are prefixed
// with dashes:
//
// lzofile ::=
//   <file-header>
//   <compressed-block>+
//
// compressed-block ::=
//   <uncompressed-size>
//   <compressed-size>
//   <uncompressed-checksums>
//   <compressed-checksums>
//   <compressed-data>
//
// file-header ::=  -- most of this information is not used.
//   <magic>
//   <version>
//   <lib-version>
//   [<version-needed>] -- present for all modern files.
//   <method>
//   <flags>
//   <mode>
//   <mtime>
//   <file-name>
//   <header-checksum>
//   <extra-field> -- presence indicated in flags, not currently used.
//
// <compressed-checksums> ::=
//   [alder-checksum | crc-checksum]
//
// <uncompressed-checksums> ::=
//   [alder-checksum | crc-checksum]
//
// <file-name> ::=
//   <length> -- one byte
//   <name>
//
namespace impala {

class ScannerContext;
class HdfsLzoTextScanner;

// HdfsScanner implementation that reads LZOP formatted text files.
// The format of the data, after decompression, is the same as HdfsText files.
// Records can span compresed blocks.
//
// An optional, but highly recommended, index file may exist in the same directory.
// This file is generated by running: com.hadoop.compression.lzo.DistributedLzoIndexer.
// The file contains the offsets to the start of each compressed block.
// This is used to find the beginning of a split and to skip over a bad block and
// find the next block.
// If there is no index file then the file is non-splittble. A single scan range
// will be issued for the whole file and no error recovery is done.


// Used to verify that this library was built against the expected Impala version when the
// library is loaded via dlopen.
extern "C" const char* GetImpalaBuildVersion() { return IMPALA_BUILD_VERSION; }

// The two functions below are wrappers for calling methods of HdfsLzoTextScanner
// when the library is loaded via dlopen.
// This function is a wrapper for the HdfsLzoTextScanner creator.  The caller is expected
// to call delete on it.
// scan_node -- scan node that is creating this scanner.
// state -- runtime state for this scanner.
extern "C" HdfsLzoTextScanner* GetLzoTextScanner(
    HdfsScanNode* scan_node, RuntimeState* state);

// This function is a wrapper for HdfsLzoTextScanner::IssueInitialRanges.
// scan_node -- scan node for this scan
// files -- files that are to be scanned.
extern "C" Status IssueInitialRanges(
    HdfsScanNode* scan_node, const std::vector<HdfsFileDesc*>& files);

class HdfsLzoTextScanner : public HdfsTextScanner {
 public:
  HdfsLzoTextScanner(HdfsScanNode* scan_node, RuntimeState* state);
  virtual ~HdfsLzoTextScanner();

  // Implementation of HdfsScanner interface not inherited from HdsfTextScanner.
  virtual Status Close();

  // This will read the header of the file, locate the index file and
  // then fire off the rest of the scan ranges.
  virtual Status ProcessSplit();

  // Issue the initial scan ranges for all lzo-text files. This reads the
  // file headers and then the reset of the file data will be issued from
  // ProcessScanRange.
  static Status IssueInitialRanges(
      HdfsScanNode* scan_node, const std::vector<HdfsFileDesc*>& files);

 private:
  enum LzoChecksum {
    CHECK_NONE,
    CHECK_CRC32,
    CHECK_ADLER
  };

  // Suffix for index files.
  const static std::string INDEX_SUFFIX;

  // Block size in bytes used by LZOP. The compressed blocks will be no bigger than this.
  const static int MAX_BLOCK_COMPRESSED_SIZE = (256 * 1024);

  // This is the fixed size of the header. It can have up to 255 bytes of 
  // file name in it as well.
  const static int MIN_HEADER_SIZE = 32;

  // An over estimate of how big the header could be.  There is a path name
  // and an option seciton.
  const static int HEADER_SIZE = 300;

  // Header informatation, shared by all scanners on this file.
  struct LzoFileHeader {
    LzoChecksum input_checksum_type_;
    LzoChecksum output_checksum_type_;

    uint32_t header_size_;

    // Offsets to compressed blocks. 
    std::vector<int64_t> offsets;
  };

  // Pointer to shared header information.
  LzoFileHeader* header_;

  // Fills the byte buffer by reading and decompressing blocks.
  virtual Status FillByteBuffer(bool* eosr, int num_bytes = 0);

  // Read header data and validate header.
  Status ReadHeader();

  // Checksum data.
  Status Checksum(LzoChecksum type,
    const std::string& source, int expected_checksum, uint8_t* buffer, int length);

  // Read the index file and set up the header.offsets.
  Status ReadIndexFile();

  // Adjust the context_ to the first block at or after the current context offset.
  Status FindFirstBlock();

  // Issue the full file ranges after reading the headers.
  Status IssueFileRanges(const char* filename);

  // Read a data block.
  // sets: byte_buffer_ptr_, byte_buffer_read_size_ and eos_read_.
  // Data will be in a mempool allocated buffer or in the disk I/O context memory
  // if the data was not compressed.
  Status ReadAndDecompressData();

  // Read compress data and recover from errosr.
  Status ReadData();

  // Pool for allocating the block_buffer_.
  boost::scoped_ptr<MemPool> block_buffer_pool_;

  // Buffer to hold decompressed data.
  uint8_t* block_buffer_;

  // Allocated length of the block_buffer_
  int32_t block_buffer_len_;

  // Next byte to be returned from the buffer holding decompressed data blocks.
  uint8_t* block_buffer_ptr_;

  // Bytes remaining in the block_buffer.
  int bytes_remaining_;

  // True if we have read the compressed block past the end of the scan.
  // If set then we return eos to the caller even if there are bytes in the buffer.
  bool past_eosr_;

  // True if the end of scan has been read.
  bool eos_read_;

  // True if we are parsing the header for this scanner.
  bool only_parsing_header_;

  // This is set when the scanner object is constructed.  Currently always true.
  // HDFS checksums the blocks from the disk to the client, so this is redundent.
  bool disable_checksum_;

  // Time spent decompressing
  RuntimeProfile::Counter* decompress_timer_;
};
}
#endif