hdfs-avro-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_AVRO_SCANNER_H
#define IMPALA_EXEC_HDFS_AVRO_SCANNER_H

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

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

#include <avro/basics.h>
#include "runtime/tuple.h"
#include "runtime/tuple-row.h"

struct avro_obj_t;
typedef struct avro_obj_t* avro_schema_t;

namespace impala {

class HdfsAvroScanner : public BaseSequenceScanner {
 public:
  static const uint8_t AVRO_VERSION_HEADER[4];

  HdfsAvroScanner(HdfsScanNode* scan_node, RuntimeState* state);

  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::AVRO;
  }

 private:
  struct ScopedAvroSchemaT {
    ScopedAvroSchemaT(avro_schema_t s = NULL) : schema(s) { }
    ScopedAvroSchemaT(const ScopedAvroSchemaT&);

    ~ScopedAvroSchemaT();

    ScopedAvroSchemaT& operator=(const ScopedAvroSchemaT&);
    ScopedAvroSchemaT& operator=(const avro_schema_t& s);

    avro_schema_t operator->() const { return schema; }
    avro_schema_t get() const { return schema; }

   private:
    avro_schema_t schema;
  };

  struct SchemaElement {
    ScopedAvroSchemaT schema;

    std::vector<SchemaElement> children;

    int null_union_position;

    const SlotDescriptor* slot_desc;
  };

  struct AvroFileHeader : public BaseSequenceScanner::FileHeader {
    std::vector<SchemaElement> schema;

    Tuple* template_tuple;

    bool use_codegend_decode_avro_data;
  };

  AvroFileHeader* avro_header_;

  static const std::string AVRO_SCHEMA_KEY;
  static const std::string AVRO_CODEC_KEY;

  static const std::string AVRO_NULL_CODEC;
  static const std::string AVRO_SNAPPY_CODEC;
  static const std::string AVRO_DEFLATE_CODEC;

  typedef int (*DecodeAvroDataFn)(HdfsAvroScanner*, int, MemPool*, uint8_t**,
                                  Tuple*, TupleRow*);

  DecodeAvroDataFn codegend_decode_avro_data_;

  Status ParseMetadata();

  Status ResolveSchemas(const avro_schema_t& table_root,
                        const avro_schema_t& file_root);

  static SchemaElement ConvertSchema(const avro_schema_t& schema);

  Status VerifyTypesMatch(SlotDescriptor* slot_desc, avro_obj_t* schema);

  int DecodeAvroData(int max_tuples, MemPool* pool, uint8_t** data,
                        Tuple* tuple, TupleRow* tuple_row);

  void MaterializeTuple(MemPool* pool, uint8_t** data, Tuple* tuple);

  static llvm::Function* CodegenDecodeAvroData(
      RuntimeState* state, llvm::Function* materialize_tuple_fn,
      const std::vector<ExprContext*>& conjunct_ctxs);

  static llvm::Function* CodegenMaterializeTuple(HdfsScanNode* node,
                                                 LlvmCodeGen* codegen);

  void ReadAvroBoolean(
      PrimitiveType type, uint8_t** data, bool write_slot, void* slot, MemPool* pool);
  void ReadAvroInt32(
      PrimitiveType type, uint8_t** data, bool write_slot, void* slot, MemPool* pool);
  void ReadAvroInt64(
      PrimitiveType type, uint8_t** data, bool write_slot, void* slot, MemPool* pool);
  void ReadAvroFloat(
      PrimitiveType type, uint8_t** data, bool write_slot, void* slot, MemPool* pool);
  void ReadAvroDouble(
      PrimitiveType type, uint8_t** data, bool write_slot, void* slot, MemPool* pool);
  void ReadAvroVarchar(
      PrimitiveType type, int max_len, uint8_t** data, bool write_slot, void* slot,
      MemPool* pool);
  void ReadAvroChar(
      PrimitiveType type, int max_len, uint8_t** data, bool write_slot, void* slot,
      MemPool* pool);
  void ReadAvroString( PrimitiveType type, uint8_t** data, bool write_slot, void* slot,
      MemPool* pool);

  void ReadAvroDecimal(
      int slot_byte_size, uint8_t** data, bool write_slot, void* slot, MemPool* pool);

  bool ReadUnionType(int null_union_position, uint8_t** data);

  static const char* LLVM_CLASS_NAME;
};
} // namespace impala

#endif // IMPALA_EXEC_HDFS_AVRO_SCANNER_H