PARQUET-428: Support INT96 and FIXED_LEN_BYTE_ARRAY types

This PR adds support for INT96 and FIXED_LEN_BYTE_ARRAY types.
It modifies the examples and DebugPrint to handle these types.

Author: Deepak Majeti <deepak.majeti@hp.com>

Closes apache#27 from majetideepak/master and squashes the following commits:

5ba0a03 [Deepak Majeti] PARQUET-428: Support INT96 and FIXED_LEN_BYTE_ARRAY types

Change-Id: I8c86350ae1245582dba6167b18f5cc92cd35e7f9

cpp/src/parquet/column_reader.cc

@@ -187,6 +187,8 @@ std::shared_ptr<ColumnReader> ColumnReader::Make(const parquet::ColumnMetaData*
  return std::make_shared<DoubleReader>(metadata, element, stream);
  case Type::BYTE_ARRAY:
  return std::make_shared<ByteArrayReader>(metadata, element, stream);
+ case Type::FIXED_LEN_BYTE_ARRAY:
+ return std::make_shared<FixedLenByteArrayReader>(metadata, element, stream);
  default:
  ParquetException::NYI("type reader not implemented");
  }

cpp/src/parquet/column_reader.h

@@ -159,6 +159,7 @@ typedef TypedColumnReader<parquet::Type::INT96> Int96Reader;
 typedef TypedColumnReader<parquet::Type::FLOAT> FloatReader;
 typedef TypedColumnReader<parquet::Type::DOUBLE> DoubleReader;
 typedef TypedColumnReader<parquet::Type::BYTE_ARRAY> ByteArrayReader;
+typedef TypedColumnReader<parquet::Type::FIXED_LEN_BYTE_ARRAY> FixedLenByteArrayReader;
 
 
 template <int TYPE>

cpp/src/parquet/encodings/dictionary-encoding.h

@@ -111,6 +111,25 @@ inline void DictionaryDecoder<parquet::Type::BYTE_ARRAY>::Init(
  }
 }
 
+template <>
+inline void DictionaryDecoder<parquet::Type::FIXED_LEN_BYTE_ARRAY>::Init(
+ Decoder<parquet::Type::FIXED_LEN_BYTE_ARRAY>* dictionary) {
+ int num_dictionary_values = dictionary->values_left();
+ dictionary_.resize(num_dictionary_values);
+ dictionary->Decode(&dictionary_[0], num_dictionary_values);
+
+ int fixed_len = schema_->type_length;
+ int total_size = num_dictionary_values*fixed_len;
+
+ byte_array_data_.resize(total_size);
+ int offset = 0;
+ for (int i = 0; i < num_dictionary_values; ++i) {
+ memcpy(&byte_array_data_[offset], dictionary_[i].ptr, fixed_len);
+ dictionary_[i].ptr = &byte_array_data_[offset];
+ offset += fixed_len;
+ }
+}
+
 } // namespace parquet_cpp
 
 #endif

cpp/src/parquet/encodings/plain-encoding.h

@@ -74,6 +74,22 @@ inline int PlainDecoder<parquet::Type::BYTE_ARRAY>::Decode(ByteArray* buffer,
  return max_values;
 }
 
+// Template specialization for FIXED_LEN_BYTE_ARRAY
+template <>
+inline int PlainDecoder<parquet::Type::FIXED_LEN_BYTE_ARRAY>::Decode(FixedLenByteArray* buffer,
+ int max_values) {
+ max_values = std::min(max_values, num_values_);
+ int len = schema_->type_length;
+ for (int i = 0; i < max_values; ++i) {
+ if (len_ < len) ParquetException::EofException();
+ buffer[i].ptr = data_;
+ data_ += len;
+ len_ -= len;
+ }
+ num_values_ -= max_values;
+ return max_values;
+}
+
 template <>
 class PlainDecoder<parquet::Type::BOOLEAN> : public Decoder<parquet::Type::BOOLEAN> {
  public:

cpp/src/parquet/reader.cc

@@ -217,6 +217,9 @@ static string parquet_type_to_string(Type::type t) {
  case Type::INT64:
  return "INT64";
  break;
+ case Type::INT96:
+ return "INT96";
+ break;
  case Type::FLOAT:
  return "FLOAT";
  break;
@@ -226,9 +229,6 @@ static string parquet_type_to_string(Type::type t) {
  case Type::BYTE_ARRAY:
  return "BYTE_ARRAY";
  break;
- case Type::INT96:
- return "INT96";
- break;
  case Type::FIXED_LEN_BYTE_ARRAY:
  return "FIXED_LEN_BYTE_ARRAY";
  break;
@@ -239,7 +239,7 @@ static string parquet_type_to_string(Type::type t) {
 }
 
 // the fixed initial size is just for an example
-#define COL_WIDTH "17"
+#define COL_WIDTH "20"
 
 void ParquetFileReader::DebugPrint(std::ostream& stream, bool print_values) {
  if (!parsed_metadata_) {
@@ -251,10 +251,6 @@ void ParquetFileReader::DebugPrint(std::ostream& stream, bool print_values) {
  for (int c = 1; c < metadata_.schema.size(); ++c) {
  stream << "Column " << c-1 << ": " << metadata_.schema[c].name << " ("
  << parquet_type_to_string(metadata_.schema[c].type);
- if (metadata_.schema[c].type == Type::INT96 ||
- metadata_.schema[c].type == Type::FIXED_LEN_BYTE_ARRAY) {
- stream << " - not supported";
- }
  stream << ")\n";
  }
 
@@ -291,10 +287,6 @@ void ParquetFileReader::DebugPrint(std::ostream& stream, bool print_values) {
 
  printf("%-" COL_WIDTH"s", metadata_.schema[c+1].name.c_str());
 
- if (col_type == Type::INT96 || col_type == Type::FIXED_LEN_BYTE_ARRAY) {
- continue;
- }
-
  // This is OK in this method as long as the RowGroupReader does not get deleted
  readers[c] = col_reader;
  }
@@ -345,6 +337,16 @@ void ParquetFileReader::DebugPrint(std::ostream& stream, bool print_values) {
  }
  break;
  }
+ case Type::INT96: {
+ Int96 val = reinterpret_cast<Int96Reader*>(readers[c])->NextValue(
+ &def_level[c], &rep_level[c]);
+ if (def_level[c] >= rep_level[c]) {
+ string result = Int96ToString(val);
+ snprintf(buffer, bufsize, "%-" COL_WIDTH"s", result.c_str());
+ stream << buffer;
+ }
+ break;
+ }
  case Type::FLOAT: {
  float val = reinterpret_cast<FloatReader*>(readers[c])->NextValue(
  &def_level[c], &rep_level[c]);
@@ -373,7 +375,17 @@ void ParquetFileReader::DebugPrint(std::ostream& stream, bool print_values) {
  }
  break;
  }
- default:
+ case Type::FIXED_LEN_BYTE_ARRAY: {
+ FixedLenByteArray val = reinterpret_cast<FixedLenByteArrayReader*>(
+ readers[c])->NextValue(&def_level[c], &rep_level[c]);
+ if (def_level[c] >= rep_level[c]) {
+ string result = FixedLenByteArrayToString(val, metadata_.schema[c+1].type_length);
+ snprintf(buffer, bufsize, "%-" COL_WIDTH"s", result.c_str());
+ stream << buffer;
+ }
+ break;
+ }
+ default:
  continue;
  }
  }

cpp/src/parquet/types.h

@@ -22,8 +22,10 @@
 #include <cstdint>
 #include <cstring>
 #include <string>
+#include <sstream>
 
 #include "parquet/thrift/parquet_types.h"
+#include "parquet/util/compiler-util.h"
 
 namespace parquet_cpp {
 
@@ -32,11 +34,36 @@ struct ByteArray {
  const uint8_t* ptr;
 };
 
+struct FixedLenByteArray {
+ const uint8_t* ptr;
+};
+
+MANUALLY_ALIGNED_STRUCT(1) Int96 {
+ uint32_t value[3];
+};
+STRUCT_END(Int96, 12);
 
 static inline std::string ByteArrayToString(const ByteArray& a) {
  return std::string(reinterpret_cast<const char*>(a.ptr), a.len);
 }
 
+static inline std::string Int96ToString(const Int96& a) {
+ std::stringstream result;
+ for (int i = 0; i < 3; i++) {
+ result << a.value[i] << " ";
+ }
+ return result.str();
+}
+
+static inline std::string FixedLenByteArrayToString(const FixedLenByteArray& a, int len) {
+ const uint8_t *bytes = reinterpret_cast<const uint8_t*>(a.ptr);
+ std::stringstream result;
+ for (int i = 0; i < len; i++) {
+ result << (uint32_t)bytes[i] << " ";
+ }
+ return result.str();
+}
+
 static inline int ByteCompare(const ByteArray& x1, const ByteArray& x2) {
  int len = std::min(x1.len, x2.len);
  int cmp = memcmp(x1.ptr, x2.ptr, len);
@@ -76,8 +103,7 @@ struct type_traits<parquet::Type::INT64> {
 
 template <>
 struct type_traits<parquet::Type::INT96> {
- // TODO
- typedef void* value_type;
+ typedef Int96 value_type;
  static constexpr parquet::Type::type parquet_type = parquet::Type::INT96;
 
  static constexpr size_t value_byte_size = 12;
@@ -107,6 +133,14 @@ struct type_traits<parquet::Type::BYTE_ARRAY> {
  static constexpr size_t value_byte_size = sizeof(ByteArray);
 };
 
+template <>
+struct type_traits<parquet::Type::FIXED_LEN_BYTE_ARRAY> {
+ typedef FixedLenByteArray value_type;
+ static constexpr parquet::Type::type parquet_type = parquet::Type::FIXED_LEN_BYTE_ARRAY;
+
+ static constexpr size_t value_byte_size = sizeof(FixedLenByteArray);
+};
+
 } // namespace parquet_cpp
 
 #endif // PARQUET_TYPES_H

cpp/src/parquet/util/compiler-util.h

@@ -36,4 +36,25 @@
 
 #define PREFETCH(addr) __builtin_prefetch(addr)
 
+//macros to disable padding
+//these macros are portable across different compilers and platforms
+//[https://github.com/google/flatbuffers/blob/master/include/flatbuffers/flatbuffers.h#L1355]
+#if defined(_MSC_VER)
+ #define MANUALLY_ALIGNED_STRUCT(alignment) \
+ __pragma(pack(1)); \
+ struct __declspec(align(alignment))
+ #define STRUCT_END(name, size) \
+ __pragma(pack()); \
+ static_assert(sizeof(name) == size, "compiler breaks packing rules")
+#elif defined(__GNUC__) || defined(__clang__)
+ #define MANUALLY_ALIGNED_STRUCT(alignment) \
+ _Pragma("pack(1)") \
+ struct __attribute__((aligned(alignment)))
+ #define STRUCT_END(name, size) \
+ _Pragma("pack()") \
+ static_assert(sizeof(name) == size, "compiler breaks packing rules")
+#else
+ #error Unknown compiler, please define structure alignment macros
+#endif
+
 #endif // PARQUET_UTIL_COMPILER_UTIL_H