PARQUET-457: Verify page deserialization for GZIP and SNAPPY codecs, …

…related refactoring

This also restores passing on user's `CMAKE_CXX_FLAGS`, which had unfortunately led some compiler warnings to creep into our build.

Author: Wes McKinney <wesm@apache.org>

Closes apache#58 from wesm/PARQUET-457 and squashes the following commits:

4bf12ed [Wes McKinney] * SerializeThriftMsg now writes into an OutputStream. * Refactor page serialization in advance of compression tests * Test compression roundtrip on random bytes for snappy and gzip * Trying LZO compression results in ParquetException * Don't lose user's CMAKE_CXX_FLAGS * Remove Travis CI directory caching for now * Fix gzip memory leak if you do not call inflateEnd, deflateEnd

Change-Id: I44a58ef2d22f8e5064d198d0abeecde7ba4de3cb

cpp/src/parquet/column/levels-test.cc

@@ -16,6 +16,7 @@
 // under the License.
 
 #include <cstdint>
+#include <memory>
 #include <vector>
 #include <string>
 

cpp/src/parquet/column/page.h

@@ -24,6 +24,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <string>
 
 #include "parquet/types.h"
 
@@ -93,13 +94,26 @@ class DataPage : public Page {
  return definition_level_encoding_;
  }
 
+ // DataPageHeader::statistics::max field, if it was set
+ const uint8_t* max() const {
+ return reinterpret_cast<const uint8_t*>(max_.c_str());
+ }
+
+ // DataPageHeader::statistics::min field, if it was set
+ const uint8_t* min() const {
+ return reinterpret_cast<const uint8_t*>(min_.c_str());
+ }
+
  private:
  int32_t num_values_;
  Encoding::type encoding_;
  Encoding::type definition_level_encoding_;
  Encoding::type repetition_level_encoding_;
 
- // TODO(wesm): parquet::DataPageHeader.statistics
+ // So max/min can be populated privately
+ friend class SerializedPageReader;
+ std::string max_;
+ std::string min_;
 };
 
 

cpp/src/parquet/column/test-util.h

@@ -32,7 +32,6 @@
 
 // Depended on by SerializedPageReader test utilities for now
 #include "parquet/encodings/plain-encoding.h"
-#include "parquet/thrift/util.h"
 #include "parquet/util/input.h"
 
 namespace parquet_cpp {
@@ -195,34 +194,6 @@ static std::shared_ptr<DataPage> MakeDataPage(const std::vector<T>& values,
 
 } // namespace test
 
-// Utilities for testing the SerializedPageReader internally
-
-static inline void InitDataPage(const parquet::Statistics& stat,
- parquet::DataPageHeader& data_page, int32_t nvalues) {
- data_page.encoding = parquet::Encoding::PLAIN;
- data_page.definition_level_encoding = parquet::Encoding::RLE;
- data_page.repetition_level_encoding = parquet::Encoding::RLE;
- data_page.num_values = nvalues;
- data_page.__set_statistics(stat);
-}
-
-static inline void InitStats(size_t stat_size, parquet::Statistics& stat) {
- std::vector<char> stat_buffer;
- stat_buffer.resize(stat_size);
- for (int i = 0; i < stat_size; i++) {
- (reinterpret_cast<uint8_t*>(stat_buffer.data()))[i] = i % 255;
- }
- stat.__set_max(std::string(stat_buffer.data(), stat_size));
-}
-
-static inline void InitPageHeader(const parquet::DataPageHeader &data_page,
- parquet::PageHeader& page_header) {
- page_header.__set_data_page_header(data_page);
- page_header.uncompressed_page_size = 0;
- page_header.compressed_page_size = 0;
- page_header.type = parquet::PageType::DATA_PAGE;
-}
-
 } // namespace parquet_cpp
 
 #endif // PARQUET_COLUMN_TEST_UTIL_H

cpp/src/parquet/compression/CMakeLists.txt

@@ -16,6 +16,7 @@
 # under the License.
 
 add_library(parquet_compression STATIC
+ codec.cc
  lz4-codec.cc
  snappy-codec.cc
  gzip-codec.cc

cpp/src/parquet/compression/codec.cc

@@ -0,0 +1,47 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+#include <memory>
+
+#include "parquet/compression/codec.h"
+#include "parquet/exception.h"
+#include "parquet/types.h"
+
+namespace parquet_cpp {
+
+std::unique_ptr<Codec> Codec::Create(Compression::type codec_type) {
+ std::unique_ptr<Codec> result;
+ switch (codec_type) {
+ case Compression::UNCOMPRESSED:
+ break;
+ case Compression::SNAPPY:
+ result.reset(new SnappyCodec());
+ break;
+ case Compression::GZIP:
+ result.reset(new GZipCodec());
+ break;
+ case Compression::LZO:
+ ParquetException::NYI("LZO codec not implemented");
+ break;
+ default:
+ ParquetException::NYI("Unrecognized codec");
+ break;
+ }
+ return result;
+}
+
+} // namespace parquet_cpp

cpp/src/parquet/compression/codec.h

@@ -19,16 +19,21 @@
 #define PARQUET_COMPRESSION_CODEC_H
 
 #include <cstdint>
+#include <memory>
 
 #include <zlib.h>
 
 #include "parquet/exception.h"
+#include "parquet/types.h"
 
 namespace parquet_cpp {
 
 class Codec {
  public:
  virtual ~Codec() {}
+
+ static std::unique_ptr<Codec> Create(Compression::type codec);
+
  virtual void Decompress(int64_t input_len, const uint8_t* input,
  int64_t output_len, uint8_t* output_buffer) = 0;
 
@@ -80,6 +85,7 @@ class GZipCodec : public Codec {
  };
 
  explicit GZipCodec(Format format = GZIP);
+ virtual ~GZipCodec();
 
  virtual void Decompress(int64_t input_len, const uint8_t* input,
  int64_t output_len, uint8_t* output_buffer);
@@ -109,6 +115,8 @@ class GZipCodec : public Codec {
  // perform the refactoring then
  void InitCompressor();
  void InitDecompressor();
+ void EndCompressor();
+ void EndDecompressor();
  bool compressor_initialized_;
  bool decompressor_initialized_;
 };

cpp/src/parquet/compression/gzip-codec.cc

@@ -15,10 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "parquet/compression/codec.h"
-
 #include <cstring>
 #include <sstream>
+#include <string>
+
+#include "parquet/compression/codec.h"
+#include "parquet/exception.h"
 
 namespace parquet_cpp {
 
@@ -40,7 +42,13 @@ GZipCodec::GZipCodec(Format format) :
  decompressor_initialized_(false) {
 }
 
+GZipCodec::~GZipCodec() {
+ EndCompressor();
+ EndDecompressor();
+}
+
 void GZipCodec::InitCompressor() {
+ EndDecompressor();
  memset(&stream_, 0, sizeof(stream_));
 
  int ret;
@@ -58,24 +66,35 @@ void GZipCodec::InitCompressor() {
  }
 
  compressor_initialized_ = true;
- decompressor_initialized_ = false;
+}
+
+void GZipCodec::EndCompressor() {
+ if (compressor_initialized_) {
+ (void)deflateEnd(&stream_);
+ }
+ compressor_initialized_ = false;
 }
 
 void GZipCodec::InitDecompressor() {
+ EndCompressor();
  memset(&stream_, 0, sizeof(stream_));
-
  int ret;
 
  // Initialize to run either deflate or zlib/gzip format
  int window_bits = format_ == DEFLATE ? -WINDOW_BITS : WINDOW_BITS | DETECT_CODEC;
  if ((ret = inflateInit2(&stream_, window_bits)) != Z_OK) {
  throw ParquetException("zlib inflateInit failed: " + std::string(stream_.msg));
  }
-
- compressor_initialized_ = false;
  decompressor_initialized_ = true;
 }
 
+void GZipCodec::EndDecompressor() {
+ if (decompressor_initialized_) {
+ (void)inflateEnd(&stream_);
+ }
+ decompressor_initialized_ = false;
+}
+
 void GZipCodec::Decompress(int64_t input_length, const uint8_t* input,
  int64_t output_length, uint8_t* output) {
  if (!decompressor_initialized_) {

cpp/src/parquet/encodings/plain-encoding-test.cc

@@ -17,11 +17,13 @@
 
 #include <cstdint>
 #include <cstdlib>
+#include <cstring>
 #include <string>
 #include <vector>
 
 #include <gtest/gtest.h>
 
+#include "parquet/schema/descriptor.h"
 #include "parquet/encodings/plain-encoding.h"
 #include "parquet/types.h"
 #include "parquet/schema/types.h"
@@ -80,7 +82,7 @@ class EncodeDecode{
 
  void generate_data() {
  // seed the prng so failure is deterministic
- random_numbers(num_values_, 0.5, draws_);
+ random_numbers(num_values_, 0, draws_);
  }
 
  void encode_decode(ColumnDescriptor *d) {
@@ -141,7 +143,7 @@ void EncodeDecode<ByteArray, Type::BYTE_ARRAY>::generate_data() {
  int max_byte_array_len = 12 + sizeof(uint32_t);
  size_t nbytes = num_values_ * max_byte_array_len;
  data_buffer_.resize(nbytes);
- random_byte_array(num_values_, 0.5, data_buffer_.data(), draws_,
+ random_byte_array(num_values_, 0, data_buffer_.data(), draws_,
  max_byte_array_len);
 }
 
@@ -160,7 +162,7 @@ void EncodeDecode<FLBA, Type::FIXED_LEN_BYTE_ARRAY>::generate_data() {
  size_t nbytes = num_values_ * flba_length;
  data_buffer_.resize(nbytes);
  ASSERT_EQ(nbytes, data_buffer_.size());
- random_fixed_byte_array(num_values_, 0.5, data_buffer_.data(), flba_length, draws_);
+ random_fixed_byte_array(num_values_, 0, data_buffer_.data(), flba_length, draws_);
 }
 
 template<>