codec.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_UTIL_CODEC_H
#define IMPALA_UTIL_CODEC_H

#include "common/status.h"
#include "runtime/mem-pool.h"
#include "util/runtime-profile.h"

#include <boost/scoped_ptr.hpp>
#include "gen-cpp/Descriptors_types.h"

namespace impala {

class MemPool;
class RuntimeState;

class Codec {
 public:
  static const char* const DEFAULT_COMPRESSION;
  static const char* const GZIP_COMPRESSION;
  static const char* const BZIP2_COMPRESSION;
  static const char* const SNAPPY_COMPRESSION;
  static const char* const UNKNOWN_CODEC_ERROR;

  typedef std::map<const std::string, const THdfsCompression::type> CodecMap;
  static const CodecMap CODEC_MAP;

  static Status CreateDecompressor(MemPool* mem_pool, bool reuse,
    THdfsCompression::type format, boost::scoped_ptr<Codec>* decompressor);

  static Status CreateDecompressor(MemPool* mem_pool, bool reuse,
      const std::string& codec, boost::scoped_ptr<Codec>* decompressor);

  static Status CreateCompressor(MemPool* mem_pool, bool reuse,
      THdfsCompression::type format, boost::scoped_ptr<Codec>* compressor);

  static Status CreateCompressor(MemPool* mem_pool, bool reuse,
      const std::string& codec, boost::scoped_ptr<Codec>* compressor);

  static std::string GetCodecName(THdfsCompression::type);
  static Status GetHadoopCodecClassName(THdfsCompression::type, std::string* out_name);

  virtual ~Codec() {}

  //
  //
  //
  virtual Status ProcessBlock(bool output_preallocated, int64_t input_length,
      const uint8_t* input, int64_t* output_length, uint8_t** output) = 0;

  Status ProcessBlock32(bool output_preallocated, int input_length, const uint8_t* input,
      int* output_length, uint8_t** output);

  virtual Status ProcessBlockStreaming(int64_t input_length, const uint8_t* input,
      int64_t* input_bytes_read, int64_t* output_length, uint8_t** output, bool* eos) {
    return Status("Not implemented.");
  }

  virtual int64_t MaxOutputLen(int64_t input_len, const uint8_t* input = NULL) = 0;

  virtual void Close();

  virtual std::string file_extension() const = 0;

  bool reuse_output_buffer() const { return reuse_buffer_; }

  static const int MAX_BLOCK_SIZE = (2L * 1024 * 1024 * 1024) - 1;

 protected:
  Codec(MemPool* mem_pool, bool reuse_buffer);

  virtual Status Init() = 0;

  MemPool* memory_pool_;

  boost::scoped_ptr<MemPool> temp_memory_pool_;

  bool reuse_buffer_;

  uint8_t* out_buffer_;

  int64_t buffer_length_;
};

}
#endif