http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/column/test-util.h ---------------------------------------------------------------------- diff --git a/src/parquet/column/test-util.h b/src/parquet/column/test-util.h index 95b1981..d13f052 100644 --- a/src/parquet/column/test-util.h +++ b/src/parquet/column/test-util.h @@ -45,15 +45,14 @@ namespace parquet { namespace test { template <typename T> -static void InitValues(int num_values, vector<T>& values, - vector<uint8_t>& buffer) { +static void InitValues(int num_values, vector<T>& values, vector<uint8_t>& buffer) { random_numbers(num_values, 0, std::numeric_limits<T>::min(), std::numeric_limits<T>::max(), values.data()); } template <typename T> -static void InitDictValues(int num_values, int num_dicts, - vector<T>& values, vector<uint8_t>& buffer) { +static void InitDictValues( + int num_values, int num_dicts, vector<T>& values, vector<uint8_t>& buffer) { int repeat_factor = num_values / num_dicts; InitValues<T>(num_dicts, values, buffer); // add some repeated values @@ -71,9 +70,8 @@ static void InitDictValues(int num_values, int num_dicts, class MockPageReader : public PageReader { public: - explicit MockPageReader(const vector<shared_ptr<Page> >& pages) : - pages_(pages), - page_index_(0) {} + explicit MockPageReader(const vector<shared_ptr<Page>>& pages) + : pages_(pages), page_index_(0) {} // Implement the PageReader interface virtual shared_ptr<Page> NextPage() { @@ -85,7 +83,7 @@ class MockPageReader : public PageReader { } private: - vector<shared_ptr<Page> > pages_; + vector<shared_ptr<Page>> pages_; int page_index_; }; @@ -97,16 +95,15 @@ class DataPageBuilder { typedef typename Type::c_type T; // This class writes data and metadata to the passed inputs - explicit DataPageBuilder(InMemoryOutputStream* sink) : - sink_(sink), - num_values_(0), - encoding_(Encoding::PLAIN), - definition_level_encoding_(Encoding::RLE), - repetition_level_encoding_(Encoding::RLE), - have_def_levels_(false), - have_rep_levels_(false), - have_values_(false) { - } + explicit DataPageBuilder(InMemoryOutputStream* sink) + : sink_(sink), + num_values_(0), + encoding_(Encoding::PLAIN), + definition_level_encoding_(Encoding::RLE), + repetition_level_encoding_(Encoding::RLE), + have_def_levels_(false), + have_rep_levels_(false), + have_values_(false) {} void AppendDefLevels(const vector<int16_t>& levels, int16_t max_level, Encoding::type encoding = Encoding::RLE) { @@ -126,7 +123,7 @@ class DataPageBuilder { have_rep_levels_ = true; } - void AppendValues(const ColumnDescriptor *d, const vector<T>& values, + void AppendValues(const ColumnDescriptor* d, const vector<T>& values, Encoding::type encoding = Encoding::PLAIN) { PlainEncoder<Type> encoder(d); encoder.Encode(&values[0], values.size(), sink_); @@ -136,21 +133,13 @@ class DataPageBuilder { have_values_ = true; } - int32_t num_values() const { - return num_values_; - } + int32_t num_values() const { return num_values_; } - Encoding::type encoding() const { - return encoding_; - } + Encoding::type encoding() const { return encoding_; } - Encoding::type rep_level_encoding() const { - return repetition_level_encoding_; - } + Encoding::type rep_level_encoding() const { return repetition_level_encoding_; } - Encoding::type def_level_encoding() const { - return definition_level_encoding_; - } + Encoding::type def_level_encoding() const { return definition_level_encoding_; } private: InMemoryOutputStream* sink_; @@ -165,8 +154,8 @@ class DataPageBuilder { bool have_values_; // Used internally for both repetition and definition levels - void AppendLevels(const vector<int16_t>& levels, int16_t max_level, - Encoding::type encoding) { + void AppendLevels( + const vector<int16_t>& levels, int16_t max_level, Encoding::type encoding) { if (encoding != Encoding::RLE) { ParquetException::NYI("only rle encoding currently implemented"); } @@ -178,8 +167,8 @@ class DataPageBuilder { // RLE-encoded bytes have to be preceded in the stream by their absolute // size. LevelEncoder encoder; - encoder.Init(encoding, max_level, levels.size(), - encode_buffer.data(), encode_buffer.size()); + encoder.Init( + encoding, max_level, levels.size(), encode_buffer.data(), encode_buffer.size()); encoder.Encode(levels.size(), levels.data()); @@ -189,9 +178,9 @@ class DataPageBuilder { } }; -template<> -void DataPageBuilder<BooleanType>::AppendValues(const ColumnDescriptor *d, - const vector<bool>& values, Encoding::type encoding) { +template <> +void DataPageBuilder<BooleanType>::AppendValues( + const ColumnDescriptor* d, const vector<bool>& values, Encoding::type encoding) { if (encoding != Encoding::PLAIN) { ParquetException::NYI("only plain encoding currently implemented"); } @@ -204,40 +193,32 @@ void DataPageBuilder<BooleanType>::AppendValues(const ColumnDescriptor *d, } template <typename Type> -static shared_ptr<DataPage> MakeDataPage(const ColumnDescriptor *d, - const vector<typename Type::c_type>& values, int num_vals, - Encoding::type encoding, const uint8_t* indices, int indices_size, - const vector<int16_t>& def_levels, int16_t max_def_level, - const vector<int16_t>& rep_levels, int16_t max_rep_level) { +static shared_ptr<DataPage> MakeDataPage(const ColumnDescriptor* d, + const vector<typename Type::c_type>& values, int num_vals, Encoding::type encoding, + const uint8_t* indices, int indices_size, const vector<int16_t>& def_levels, + int16_t max_def_level, const vector<int16_t>& rep_levels, int16_t max_rep_level) { int num_values = 0; InMemoryOutputStream page_stream; test::DataPageBuilder<Type> page_builder(&page_stream); - if (!rep_levels.empty()) { - page_builder.AppendRepLevels(rep_levels, max_rep_level); - } - if (!def_levels.empty()) { - page_builder.AppendDefLevels(def_levels, max_def_level); - } + if (!rep_levels.empty()) { page_builder.AppendRepLevels(rep_levels, max_rep_level); } + if (!def_levels.empty()) { page_builder.AppendDefLevels(def_levels, max_def_level); } if (encoding == Encoding::PLAIN) { page_builder.AppendValues(d, values, encoding); num_values = page_builder.num_values(); - } else {// DICTIONARY PAGES + } else { // DICTIONARY PAGES page_stream.Write(indices, indices_size); num_values = std::max(page_builder.num_values(), num_vals); } auto buffer = page_stream.GetBuffer(); - return std::make_shared<DataPage>(buffer, num_values, - encoding, - page_builder.def_level_encoding(), - page_builder.rep_level_encoding()); + return std::make_shared<DataPage>(buffer, num_values, encoding, + page_builder.def_level_encoding(), page_builder.rep_level_encoding()); } - template <typename TYPE> class DictionaryPageBuilder { public: @@ -245,19 +226,14 @@ class DictionaryPageBuilder { static constexpr int TN = TYPE::type_num; // This class writes data and metadata to the passed inputs - explicit DictionaryPageBuilder(const ColumnDescriptor *d) : - num_dict_values_(0), - have_values_(false) { - int type_length = 0; - if (TN == Type::FIXED_LEN_BYTE_ARRAY) { - type_length = d->type_length(); - } - encoder_.reset(new DictEncoder<TC>(&pool_, default_allocator(), type_length)); + explicit DictionaryPageBuilder(const ColumnDescriptor* d) + : num_dict_values_(0), have_values_(false) { + int type_length = 0; + if (TN == Type::FIXED_LEN_BYTE_ARRAY) { type_length = d->type_length(); } + encoder_.reset(new DictEncoder<TC>(&pool_, default_allocator(), type_length)); } - ~DictionaryPageBuilder() { - pool_.FreeAll(); - } + ~DictionaryPageBuilder() { pool_.FreeAll(); } shared_ptr<Buffer> AppendValues(const vector<TC>& values) { int num_values = values.size(); @@ -269,43 +245,41 @@ class DictionaryPageBuilder { have_values_ = true; shared_ptr<OwnedMutableBuffer> rle_indices = std::make_shared<OwnedMutableBuffer>( sizeof(int) * encoder_->EstimatedDataEncodedSize()); - int actual_bytes = encoder_->WriteIndices(rle_indices->mutable_data(), - rle_indices->size()); + int actual_bytes = + encoder_->WriteIndices(rle_indices->mutable_data(), rle_indices->size()); rle_indices->Resize(actual_bytes); encoder_->ClearIndices(); return rle_indices; } shared_ptr<Buffer> WriteDict() { - shared_ptr<OwnedMutableBuffer> dict_buffer = std::make_shared<OwnedMutableBuffer>( - encoder_->dict_encoded_size()); + shared_ptr<OwnedMutableBuffer> dict_buffer = + std::make_shared<OwnedMutableBuffer>(encoder_->dict_encoded_size()); encoder_->WriteDict(dict_buffer->mutable_data()); return dict_buffer; } - int32_t num_values() const { - return num_dict_values_; - } + int32_t num_values() const { return num_dict_values_; } private: MemPool pool_; - shared_ptr<DictEncoder<TC> > encoder_; + shared_ptr<DictEncoder<TC>> encoder_; int32_t num_dict_values_; bool have_values_; }; -template<> -DictionaryPageBuilder<BooleanType>::DictionaryPageBuilder(const ColumnDescriptor *d) { +template <> +DictionaryPageBuilder<BooleanType>::DictionaryPageBuilder(const ColumnDescriptor* d) { ParquetException::NYI("only plain encoding currently implemented for boolean"); } -template<> +template <> shared_ptr<Buffer> DictionaryPageBuilder<BooleanType>::WriteDict() { ParquetException::NYI("only plain encoding currently implemented for boolean"); return nullptr; } -template<> +template <> shared_ptr<Buffer> DictionaryPageBuilder<BooleanType>::AppendValues( const vector<TC>& values) { ParquetException::NYI("only plain encoding currently implemented for boolean"); @@ -313,39 +287,37 @@ shared_ptr<Buffer> DictionaryPageBuilder<BooleanType>::AppendValues( } template <typename Type> -static shared_ptr<DictionaryPage> MakeDictPage(const ColumnDescriptor *d, +static shared_ptr<DictionaryPage> MakeDictPage(const ColumnDescriptor* d, const vector<typename Type::c_type>& values, const vector<int>& values_per_page, - Encoding::type encoding, vector<shared_ptr<Buffer> >& rle_indices) { + Encoding::type encoding, vector<shared_ptr<Buffer>>& rle_indices) { InMemoryOutputStream page_stream; test::DictionaryPageBuilder<Type> page_builder(d); int num_pages = values_per_page.size(); int value_start = 0; for (int i = 0; i < num_pages; i++) { - rle_indices.push_back(page_builder.AppendValues(slice(values, value_start, - value_start + values_per_page[i]))); + rle_indices.push_back(page_builder.AppendValues( + slice(values, value_start, value_start + values_per_page[i]))); value_start += values_per_page[i]; } auto buffer = page_builder.WriteDict(); - return std::make_shared<DictionaryPage>(buffer, page_builder.num_values(), - Encoding::PLAIN); + return std::make_shared<DictionaryPage>( + buffer, page_builder.num_values(), Encoding::PLAIN); } // Given def/rep levels and values create multiple dict pages template <typename Type> -static void PaginateDict(const ColumnDescriptor *d, - const vector<typename Type::c_type>& values, - const vector<int16_t>& def_levels, int16_t max_def_level, - const vector<int16_t>& rep_levels, int16_t max_rep_level, +static void PaginateDict(const ColumnDescriptor* d, + const vector<typename Type::c_type>& values, const vector<int16_t>& def_levels, + int16_t max_def_level, const vector<int16_t>& rep_levels, int16_t max_rep_level, int num_levels_per_page, const vector<int>& values_per_page, - vector<shared_ptr<Page> >& pages, - Encoding::type encoding = Encoding::RLE_DICTIONARY) { + vector<shared_ptr<Page>>& pages, Encoding::type encoding = Encoding::RLE_DICTIONARY) { int num_pages = values_per_page.size(); - vector<shared_ptr<Buffer> > rle_indices; - shared_ptr<DictionaryPage> dict_page = MakeDictPage<Type>(d, values, values_per_page, - encoding, rle_indices); + vector<shared_ptr<Buffer>> rle_indices; + shared_ptr<DictionaryPage> dict_page = + MakeDictPage<Type>(d, values, values_per_page, encoding, rle_indices); pages.push_back(dict_page); int def_level_start = 0; int def_level_end = 0; @@ -370,13 +342,11 @@ static void PaginateDict(const ColumnDescriptor *d, // Given def/rep levels and values create multiple plain pages template <typename Type> -static void PaginatePlain(const ColumnDescriptor *d, - const vector<typename Type::c_type>& values, - const vector<int16_t>& def_levels, int16_t max_def_level, - const vector<int16_t>& rep_levels, int16_t max_rep_level, +static void PaginatePlain(const ColumnDescriptor* d, + const vector<typename Type::c_type>& values, const vector<int16_t>& def_levels, + int16_t max_def_level, const vector<int16_t>& rep_levels, int16_t max_rep_level, int num_levels_per_page, const vector<int>& values_per_page, - vector<shared_ptr<Page> >& pages, - Encoding::type encoding = Encoding::PLAIN) { + vector<shared_ptr<Page>>& pages, Encoding::type encoding = Encoding::PLAIN) { int num_pages = values_per_page.size(); int def_level_start = 0; int def_level_end = 0; @@ -394,9 +364,8 @@ static void PaginatePlain(const ColumnDescriptor *d, } shared_ptr<DataPage> page = MakeDataPage<Type>(d, slice(values, value_start, value_start + values_per_page[i]), values_per_page[i], - encoding, NULL, 0, - slice(def_levels, def_level_start, def_level_end), max_def_level, - slice(rep_levels, rep_level_start, rep_level_end), max_rep_level); + encoding, NULL, 0, slice(def_levels, def_level_start, def_level_end), + max_def_level, slice(rep_levels, rep_level_start, rep_level_end), max_rep_level); pages.push_back(page); value_start += values_per_page[i]; } @@ -404,11 +373,10 @@ static void PaginatePlain(const ColumnDescriptor *d, // Generates pages from randomly generated data template <typename Type> -static int MakePages(const ColumnDescriptor *d, int num_pages, int levels_per_page, +static int MakePages(const ColumnDescriptor* d, int num_pages, int levels_per_page, vector<int16_t>& def_levels, vector<int16_t>& rep_levels, vector<typename Type::c_type>& values, vector<uint8_t>& buffer, - vector<shared_ptr<Page> >& pages, - Encoding::type encoding = Encoding::PLAIN) { + vector<shared_ptr<Page>>& pages, Encoding::type encoding = Encoding::PLAIN) { int num_levels = levels_per_page * num_pages; int num_values = 0; uint32_t seed = 0; @@ -442,21 +410,21 @@ static int MakePages(const ColumnDescriptor *d, int num_pages, int levels_per_pa values.resize(num_values); if (encoding == Encoding::PLAIN) { InitValues<typename Type::c_type>(num_values, values, buffer); - PaginatePlain<Type>(d, values, def_levels, max_def_level, - rep_levels, max_rep_level, levels_per_page, values_per_page, pages); - } else if (encoding == Encoding::RLE_DICTIONARY - || encoding == Encoding::PLAIN_DICTIONARY) { + PaginatePlain<Type>(d, values, def_levels, max_def_level, rep_levels, max_rep_level, + levels_per_page, values_per_page, pages); + } else if (encoding == Encoding::RLE_DICTIONARY || + encoding == Encoding::PLAIN_DICTIONARY) { // Calls InitValues and repeats the data InitDictValues<typename Type::c_type>(num_values, levels_per_page, values, buffer); - PaginateDict<Type>(d, values, def_levels, max_def_level, - rep_levels, max_rep_level, levels_per_page, values_per_page, pages); + PaginateDict<Type>(d, values, def_levels, max_def_level, rep_levels, max_rep_level, + levels_per_page, values_per_page, pages); } return num_values; } -} // namespace test +} // namespace test -} // namespace parquet +} // namespace parquet -#endif // PARQUET_COLUMN_TEST_UTIL_H +#endif // PARQUET_COLUMN_TEST_UTIL_H
http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/column/writer.cc ---------------------------------------------------------------------- diff --git a/src/parquet/column/writer.cc b/src/parquet/column/writer.cc index 4dcb672..584b8a7 100644 --- a/src/parquet/column/writer.cc +++ b/src/parquet/column/writer.cc @@ -25,12 +25,15 @@ namespace parquet { // ColumnWriter ColumnWriter::ColumnWriter(const ColumnDescriptor* descr, - std::unique_ptr<PageWriter> pager, int64_t expected_rows, - MemoryAllocator* allocator): - descr_(descr), pager_(std::move(pager)), expected_rows_(expected_rows), - allocator_(allocator), - num_buffered_values_(0), num_buffered_encoded_values_(0), - num_rows_(0), total_bytes_written_(0) { + std::unique_ptr<PageWriter> pager, int64_t expected_rows, MemoryAllocator* allocator) + : descr_(descr), + pager_(std::move(pager)), + expected_rows_(expected_rows), + allocator_(allocator), + num_buffered_values_(0), + num_buffered_encoded_values_(0), + num_rows_(0), + total_bytes_written_(0) { InitSinks(); } @@ -41,13 +44,13 @@ void ColumnWriter::InitSinks() { } void ColumnWriter::WriteDefinitionLevels(int64_t num_levels, int16_t* levels) { - definition_levels_sink_->Write(reinterpret_cast<uint8_t*>(levels), - sizeof(int16_t) * num_levels); + definition_levels_sink_->Write( + reinterpret_cast<uint8_t*>(levels), sizeof(int16_t) * num_levels); } void ColumnWriter::WriteRepetitionLevels(int64_t num_levels, int16_t* levels) { - repetition_levels_sink_->Write(reinterpret_cast<uint8_t*>(levels), - sizeof(int16_t) * num_levels); + repetition_levels_sink_->Write( + reinterpret_cast<uint8_t*>(levels), sizeof(int16_t) * num_levels); } std::shared_ptr<Buffer> ColumnWriter::RleEncodeLevels( @@ -55,11 +58,11 @@ std::shared_ptr<Buffer> ColumnWriter::RleEncodeLevels( // TODO: This only works with due to some RLE specifics int64_t rle_size = 2 * num_buffered_values_ + sizeof(uint32_t); auto buffer_rle = std::make_shared<OwnedMutableBuffer>(rle_size, allocator_); - level_encoder_.Init(Encoding::RLE, max_level, - num_buffered_values_, buffer_rle->mutable_data() + sizeof(uint32_t), + level_encoder_.Init(Encoding::RLE, max_level, num_buffered_values_, + buffer_rle->mutable_data() + sizeof(uint32_t), buffer_rle->size() - sizeof(uint32_t)); - int encoded = level_encoder_.Encode(num_buffered_values_, - reinterpret_cast<const int16_t*>(buffer->data())); + int encoded = level_encoder_.Encode( + num_buffered_values_, reinterpret_cast<const int16_t*>(buffer->data())); DCHECK_EQ(encoded, num_buffered_values_); reinterpret_cast<uint32_t*>(buffer_rle->mutable_data())[0] = level_encoder_.len(); int64_t encoded_size = level_encoder_.len() + sizeof(uint32_t); @@ -75,21 +78,19 @@ void ColumnWriter::WriteNewPage() { std::shared_ptr<Buffer> values = values_sink_->GetBuffer(); if (descr_->max_definition_level() > 0) { - definition_levels = RleEncodeLevels(definition_levels, - descr_->max_definition_level()); + definition_levels = + RleEncodeLevels(definition_levels, descr_->max_definition_level()); } if (descr_->max_repetition_level() > 0) { - repetition_levels = RleEncodeLevels(repetition_levels, - descr_->max_repetition_level()); + repetition_levels = + RleEncodeLevels(repetition_levels, descr_->max_repetition_level()); } // TODO(PARQUET-590): Encodings are hard-coded int64_t bytes_written = pager_->WriteDataPage(num_buffered_values_, - num_buffered_encoded_values_, - definition_levels, Encoding::RLE, - repetition_levels, Encoding::RLE, - values, Encoding::PLAIN); + num_buffered_encoded_values_, definition_levels, Encoding::RLE, repetition_levels, + Encoding::RLE, values, Encoding::PLAIN); total_bytes_written_ += bytes_written; // Re-initialize the sinks as GetBuffer made them invalid. @@ -100,12 +101,11 @@ void ColumnWriter::WriteNewPage() { int64_t ColumnWriter::Close() { // Write all outstanding data to a new page - if (num_buffered_values_ > 0) { - WriteNewPage(); - } + if (num_buffered_values_ > 0) { WriteNewPage(); } if (num_rows_ != expected_rows_) { - throw ParquetException("Less then the number of expected rows written in" + throw ParquetException( + "Less then the number of expected rows written in" " the current column chunk"); } @@ -119,47 +119,44 @@ int64_t ColumnWriter::Close() { template <typename Type> TypedColumnWriter<Type>::TypedColumnWriter(const ColumnDescriptor* schema, - std::unique_ptr<PageWriter> pager, int64_t expected_rows, - MemoryAllocator* allocator) : - ColumnWriter(schema, std::move(pager), expected_rows, allocator) { + std::unique_ptr<PageWriter> pager, int64_t expected_rows, MemoryAllocator* allocator) + : ColumnWriter(schema, std::move(pager), expected_rows, allocator) { // TODO(PARQUET-590) Get decoder type from WriterProperties - current_encoder_ = std::unique_ptr<EncoderType>( - new PlainEncoder<Type>(schema, allocator)); + current_encoder_ = + std::unique_ptr<EncoderType>(new PlainEncoder<Type>(schema, allocator)); } // ---------------------------------------------------------------------- // Dynamic column writer constructor -std::shared_ptr<ColumnWriter> ColumnWriter::Make( - const ColumnDescriptor* descr, - std::unique_ptr<PageWriter> pager, - int64_t expected_rows, +std::shared_ptr<ColumnWriter> ColumnWriter::Make(const ColumnDescriptor* descr, + std::unique_ptr<PageWriter> pager, int64_t expected_rows, MemoryAllocator* allocator) { switch (descr->physical_type()) { case Type::BOOLEAN: - return std::make_shared<BoolWriter>(descr, std::move(pager), expected_rows, - allocator); + return std::make_shared<BoolWriter>( + descr, std::move(pager), expected_rows, allocator); case Type::INT32: - return std::make_shared<Int32Writer>(descr, std::move(pager), expected_rows, - allocator); + return std::make_shared<Int32Writer>( + descr, std::move(pager), expected_rows, allocator); case Type::INT64: - return std::make_shared<Int64Writer>(descr, std::move(pager), expected_rows, - allocator); + return std::make_shared<Int64Writer>( + descr, std::move(pager), expected_rows, allocator); case Type::INT96: - return std::make_shared<Int96Writer>(descr, std::move(pager), expected_rows, - allocator); + return std::make_shared<Int96Writer>( + descr, std::move(pager), expected_rows, allocator); case Type::FLOAT: - return std::make_shared<FloatWriter>(descr, std::move(pager), expected_rows, - allocator); + return std::make_shared<FloatWriter>( + descr, std::move(pager), expected_rows, allocator); case Type::DOUBLE: - return std::make_shared<DoubleWriter>(descr, std::move(pager), expected_rows, - allocator); + return std::make_shared<DoubleWriter>( + descr, std::move(pager), expected_rows, allocator); case Type::BYTE_ARRAY: - return std::make_shared<ByteArrayWriter>(descr, std::move(pager), expected_rows, - allocator); + return std::make_shared<ByteArrayWriter>( + descr, std::move(pager), expected_rows, allocator); case Type::FIXED_LEN_BYTE_ARRAY: - return std::make_shared<FixedLenByteArrayWriter>(descr, - std::move(pager), expected_rows, allocator); + return std::make_shared<FixedLenByteArrayWriter>( + descr, std::move(pager), expected_rows, allocator); default: ParquetException::NYI("type reader not implemented"); } @@ -179,4 +176,4 @@ template class TypedColumnWriter<DoubleType>; template class TypedColumnWriter<ByteArrayType>; template class TypedColumnWriter<FLBAType>; -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/column/writer.h ---------------------------------------------------------------------- diff --git a/src/parquet/column/writer.h b/src/parquet/column/writer.h index 7ccfe73..32d0712 100644 --- a/src/parquet/column/writer.h +++ b/src/parquet/column/writer.h @@ -37,13 +37,9 @@ class ColumnWriter { std::unique_ptr<PageWriter>, int64_t expected_rows, MemoryAllocator* allocator = default_allocator()); - Type::type type() const { - return descr_->physical_type(); - } + Type::type type() const { return descr_->physical_type(); } - const ColumnDescriptor* descr() const { - return descr_; - } + const ColumnDescriptor* descr() const { return descr_; } /** * Closes the ColumnWriter, commits any buffered values to pages. @@ -61,8 +57,8 @@ class ColumnWriter { // Write multiple repetition levels void WriteRepetitionLevels(int64_t num_levels, int16_t* levels); - std::shared_ptr<Buffer> RleEncodeLevels(const std::shared_ptr<Buffer>& buffer, - int16_t max_level); + std::shared_ptr<Buffer> RleEncodeLevels( + const std::shared_ptr<Buffer>& buffer, int16_t max_level); const ColumnDescriptor* descr_; @@ -106,14 +102,13 @@ class TypedColumnWriter : public ColumnWriter { public: typedef typename DType::c_type T; - TypedColumnWriter(const ColumnDescriptor* schema, - std::unique_ptr<PageWriter> pager, int64_t expected_rows, - MemoryAllocator* allocator = default_allocator()); + TypedColumnWriter(const ColumnDescriptor* schema, std::unique_ptr<PageWriter> pager, + int64_t expected_rows, MemoryAllocator* allocator = default_allocator()); // Write a batch of repetition levels, definition levels, and values to the // column. - void WriteBatch(int64_t num_values, int16_t* def_levels, int16_t* rep_levels, - T* values); + void WriteBatch( + int64_t num_values, int16_t* def_levels, int16_t* rep_levels, T* values); private: typedef Encoder<DType> EncoderType; @@ -124,7 +119,7 @@ class TypedColumnWriter : public ColumnWriter { // Map of encoding type to the respective encoder object. For example, a // column chunk's data pages may include both dictionary-encoded and // plain-encoded data. - std::unordered_map<int, std::shared_ptr<EncoderType> > encoders_; + std::unordered_map<int, std::shared_ptr<EncoderType>> encoders_; void ConfigureDictionary(const DictionaryPage* page); @@ -136,16 +131,14 @@ class TypedColumnWriter : public ColumnWriter { const int64_t PAGE_VALUE_COUNT = 1000; template <typename DType> -inline void TypedColumnWriter<DType>::WriteBatch(int64_t num_values, int16_t* def_levels, - int16_t* rep_levels, T* values) { +inline void TypedColumnWriter<DType>::WriteBatch( + int64_t num_values, int16_t* def_levels, int16_t* rep_levels, T* values) { int64_t values_to_write = 0; // If the field is required and non-repeated, there are no definition levels if (descr_->max_definition_level() > 0) { for (int64_t i = 0; i < num_values; ++i) { - if (def_levels[i] == descr_->max_definition_level()) { - ++values_to_write; - } + if (def_levels[i] == descr_->max_definition_level()) { ++values_to_write; } } WriteDefinitionLevels(num_values, def_levels); @@ -159,9 +152,7 @@ inline void TypedColumnWriter<DType>::WriteBatch(int64_t num_values, int16_t* de // A row could include more than one value // Count the occasions where we start a new row for (int64_t i = 0; i < num_values; ++i) { - if (rep_levels[i] == 0) { - num_rows_++; - } + if (rep_levels[i] == 0) { num_rows_++; } } WriteRepetitionLevels(num_values, rep_levels); @@ -180,9 +171,7 @@ inline void TypedColumnWriter<DType>::WriteBatch(int64_t num_values, int16_t* de num_buffered_encoded_values_ += values_to_write; // TODO(PARQUET-591): Instead of rows as a boundary, do a size check - if (num_buffered_values_ >= PAGE_VALUE_COUNT) { - WriteNewPage(); - } + if (num_buffered_values_ >= PAGE_VALUE_COUNT) { WriteNewPage(); } } template <typename DType> @@ -199,6 +188,6 @@ typedef TypedColumnWriter<DoubleType> DoubleWriter; typedef TypedColumnWriter<ByteArrayType> ByteArrayWriter; typedef TypedColumnWriter<FLBAType> FixedLenByteArrayWriter; -} // namespace parquet +} // namespace parquet -#endif // PARQUET_COLUMN_READER_H +#endif // PARQUET_COLUMN_READER_H http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/compression/codec-test.cc ---------------------------------------------------------------------- diff --git a/src/parquet/compression/codec-test.cc b/src/parquet/compression/codec-test.cc index 6326003..d6ddbd6 100644 --- a/src/parquet/compression/codec-test.cc +++ b/src/parquet/compression/codec-test.cc @@ -39,24 +39,22 @@ void CheckCodecRoundtrip(const vector<uint8_t>& data) { std::vector<uint8_t> decompressed(data.size()); // compress with c1 - int actual_size = c1.Compress(data.size(), &data[0], max_compressed_len, - &compressed[0]); + int actual_size = + c1.Compress(data.size(), &data[0], max_compressed_len, &compressed[0]); compressed.resize(actual_size); // decompress with c2 - c2.Decompress(compressed.size(), &compressed[0], - decompressed.size(), &decompressed[0]); + c2.Decompress(compressed.size(), &compressed[0], decompressed.size(), &decompressed[0]); ASSERT_TRUE(test::vector_equal(data, decompressed)); // compress with c2 - int actual_size2 = c2.Compress(data.size(), &data[0], max_compressed_len, - &compressed[0]); + int actual_size2 = + c2.Compress(data.size(), &data[0], max_compressed_len, &compressed[0]); ASSERT_EQ(actual_size2, actual_size); // decompress with c1 - c1.Decompress(compressed.size(), &compressed[0], - decompressed.size(), &decompressed[0]); + c1.Decompress(compressed.size(), &compressed[0], decompressed.size(), &decompressed[0]); ASSERT_TRUE(test::vector_equal(data, decompressed)); } @@ -83,4 +81,4 @@ TEST(TestCompressors, GZip) { CheckCodec<GZipCodec>(); } -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/compression/codec.cc ---------------------------------------------------------------------- diff --git a/src/parquet/compression/codec.cc b/src/parquet/compression/codec.cc index 2023fcd..fed5644 100644 --- a/src/parquet/compression/codec.cc +++ b/src/parquet/compression/codec.cc @@ -44,4 +44,4 @@ std::unique_ptr<Codec> Codec::Create(Compression::type codec_type) { return result; } -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/compression/codec.h ---------------------------------------------------------------------- diff --git a/src/parquet/compression/codec.h b/src/parquet/compression/codec.h index 95d6014..ffbe563 100644 --- a/src/parquet/compression/codec.h +++ b/src/parquet/compression/codec.h @@ -34,8 +34,8 @@ class 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; + virtual void Decompress(int64_t input_len, const uint8_t* input, int64_t output_len, + uint8_t* output_buffer) = 0; virtual int64_t Compress(int64_t input_len, const uint8_t* input, int64_t output_buffer_len, uint8_t* output_buffer) = 0; @@ -45,12 +45,11 @@ class Codec { virtual const char* name() const = 0; }; - // Snappy codec. class SnappyCodec : public Codec { public: - virtual void Decompress(int64_t input_len, const uint8_t* input, - int64_t output_len, uint8_t* output_buffer); + virtual void Decompress(int64_t input_len, const uint8_t* input, int64_t output_len, + uint8_t* output_buffer); virtual int64_t Compress(int64_t input_len, const uint8_t* input, int64_t output_buffer_len, uint8_t* output_buffer); @@ -63,8 +62,8 @@ class SnappyCodec : public Codec { // Lz4 codec. class Lz4Codec : public Codec { public: - virtual void Decompress(int64_t input_len, const uint8_t* input, - int64_t output_len, uint8_t* output_buffer); + virtual void Decompress(int64_t input_len, const uint8_t* input, int64_t output_len, + uint8_t* output_buffer); virtual int64_t Compress(int64_t input_len, const uint8_t* input, int64_t output_buffer_len, uint8_t* output_buffer); @@ -87,8 +86,8 @@ 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); + virtual void Decompress(int64_t input_len, const uint8_t* input, int64_t output_len, + uint8_t* output_buffer); virtual int64_t Compress(int64_t input_len, const uint8_t* input, int64_t output_buffer_len, uint8_t* output_buffer); @@ -121,6 +120,6 @@ class GZipCodec : public Codec { bool decompressor_initialized_; }; -} // namespace parquet +} // namespace parquet #endif http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/compression/gzip-codec.cc ---------------------------------------------------------------------- diff --git a/src/parquet/compression/gzip-codec.cc b/src/parquet/compression/gzip-codec.cc index b71afc3..6c27714 100644 --- a/src/parquet/compression/gzip-codec.cc +++ b/src/parquet/compression/gzip-codec.cc @@ -36,11 +36,8 @@ static constexpr int GZIP_CODEC = 16; // Determine if this is libz or gzip from header. static constexpr int DETECT_CODEC = 32; -GZipCodec::GZipCodec(Format format) : - format_(format), - compressor_initialized_(false), - decompressor_initialized_(false) { -} +GZipCodec::GZipCodec(Format format) + : format_(format), compressor_initialized_(false), decompressor_initialized_(false) {} GZipCodec::~GZipCodec() { EndCompressor(); @@ -59,19 +56,16 @@ void GZipCodec::InitCompressor() { } else if (format_ == GZIP) { window_bits += GZIP_CODEC; } - if ((ret = deflateInit2(&stream_, Z_DEFAULT_COMPRESSION, Z_DEFLATED, - window_bits, 9, Z_DEFAULT_STRATEGY)) != Z_OK) { - throw ParquetException("zlib deflateInit failed: " + - std::string(stream_.msg)); + if ((ret = deflateInit2(&stream_, Z_DEFAULT_COMPRESSION, Z_DEFLATED, window_bits, 9, + Z_DEFAULT_STRATEGY)) != Z_OK) { + throw ParquetException("zlib deflateInit failed: " + std::string(stream_.msg)); } compressor_initialized_ = true; } void GZipCodec::EndCompressor() { - if (compressor_initialized_) { - (void)deflateEnd(&stream_); - } + if (compressor_initialized_) { (void)deflateEnd(&stream_); } compressor_initialized_ = false; } @@ -83,23 +77,19 @@ void GZipCodec::InitDecompressor() { // 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)); + throw ParquetException("zlib inflateInit failed: " + std::string(stream_.msg)); } decompressor_initialized_ = true; } void GZipCodec::EndDecompressor() { - if (decompressor_initialized_) { - (void)inflateEnd(&stream_); - } + 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_) { - InitDecompressor(); - } +void GZipCodec::Decompress( + int64_t input_length, const uint8_t* input, int64_t output_length, uint8_t* output) { + if (!decompressor_initialized_) { InitDecompressor(); } if (output_length == 0) { // The zlib library does not allow *output to be NULL, even when output_length // is 0 (inflate() will return Z_STREAM_ERROR). We don't consider this an @@ -133,8 +123,8 @@ void GZipCodec::Decompress(int64_t input_length, const uint8_t* input, // Failure, buffer was too small std::stringstream ss; - ss << "Too small a buffer passed to GZipCodec. InputLength=" - << input_length << " OutputLength=" << output_length; + ss << "Too small a buffer passed to GZipCodec. InputLength=" << input_length + << " OutputLength=" << output_length; throw ParquetException(ss.str()); } @@ -149,18 +139,14 @@ void GZipCodec::Decompress(int64_t input_length, const uint8_t* input, int64_t GZipCodec::MaxCompressedLen(int64_t input_length, const uint8_t* input) { // Most be in compression mode - if (!compressor_initialized_) { - InitCompressor(); - } + if (!compressor_initialized_) { InitCompressor(); } // TODO(wesm): deal with zlib < 1.2.3 (see Impala codebase) return deflateBound(&stream_, static_cast<uLong>(input_length)); } -int64_t GZipCodec::Compress(int64_t input_length, const uint8_t* input, - int64_t output_length, uint8_t* output) { - if (!compressor_initialized_) { - InitCompressor(); - } +int64_t GZipCodec::Compress( + int64_t input_length, const uint8_t* input, int64_t output_length, uint8_t* output) { + if (!compressor_initialized_) { InitCompressor(); } stream_.next_in = const_cast<Bytef*>(reinterpret_cast<const Bytef*>(input)); stream_.avail_in = input_length; stream_.next_out = reinterpret_cast<Bytef*>(output); @@ -179,12 +165,11 @@ int64_t GZipCodec::Compress(int64_t input_length, const uint8_t* input, } if (deflateReset(&stream_) != Z_OK) { - throw ParquetException("zlib deflateReset failed: " + - std::string(stream_.msg)); + throw ParquetException("zlib deflateReset failed: " + std::string(stream_.msg)); } // Actual output length return output_length - stream_.avail_out; } -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/compression/lz4-codec.cc ---------------------------------------------------------------------- diff --git a/src/parquet/compression/lz4-codec.cc b/src/parquet/compression/lz4-codec.cc index cc91bf0..7acc1de 100644 --- a/src/parquet/compression/lz4-codec.cc +++ b/src/parquet/compression/lz4-codec.cc @@ -23,13 +23,11 @@ namespace parquet { -void Lz4Codec::Decompress(int64_t input_len, const uint8_t* input, - int64_t output_len, uint8_t* output_buffer) { +void Lz4Codec::Decompress( + int64_t input_len, const uint8_t* input, int64_t output_len, uint8_t* output_buffer) { int64_t n = LZ4_decompress_fast(reinterpret_cast<const char*>(input), reinterpret_cast<char*>(output_buffer), output_len); - if (n != input_len) { - throw ParquetException("Corrupt lz4 compressed data."); - } + if (n != input_len) { throw ParquetException("Corrupt lz4 compressed data."); } } int64_t Lz4Codec::MaxCompressedLen(int64_t input_len, const uint8_t* input) { @@ -42,4 +40,4 @@ int64_t Lz4Codec::Compress(int64_t input_len, const uint8_t* input, reinterpret_cast<char*>(output_buffer), input_len); } -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/compression/snappy-codec.cc ---------------------------------------------------------------------- diff --git a/src/parquet/compression/snappy-codec.cc b/src/parquet/compression/snappy-codec.cc index 62ba00a..ccd25b9 100644 --- a/src/parquet/compression/snappy-codec.cc +++ b/src/parquet/compression/snappy-codec.cc @@ -24,10 +24,10 @@ namespace parquet { -void SnappyCodec::Decompress(int64_t input_len, const uint8_t* input, - int64_t output_len, uint8_t* output_buffer) { +void SnappyCodec::Decompress( + int64_t input_len, const uint8_t* input, int64_t output_len, uint8_t* output_buffer) { if (!snappy::RawUncompress(reinterpret_cast<const char*>(input), - static_cast<size_t>(input_len), reinterpret_cast<char*>(output_buffer))) { + static_cast<size_t>(input_len), reinterpret_cast<char*>(output_buffer))) { throw parquet::ParquetException("Corrupt snappy compressed data."); } } @@ -45,4 +45,4 @@ int64_t SnappyCodec::Compress(int64_t input_len, const uint8_t* input, return output_len; } -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/encodings/decoder.h ---------------------------------------------------------------------- diff --git a/src/parquet/encodings/decoder.h b/src/parquet/encodings/decoder.h index 36af107..4442507 100644 --- a/src/parquet/encodings/decoder.h +++ b/src/parquet/encodings/decoder.h @@ -65,6 +65,6 @@ class Decoder { int num_values_; }; -} // namespace parquet +} // namespace parquet -#endif // PARQUET_ENCODINGS_DECODER_H +#endif // PARQUET_ENCODINGS_DECODER_H http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/encodings/delta-bit-pack-encoding.h ---------------------------------------------------------------------- diff --git a/src/parquet/encodings/delta-bit-pack-encoding.h b/src/parquet/encodings/delta-bit-pack-encoding.h index b0a16a7..5353817 100644 --- a/src/parquet/encodings/delta-bit-pack-encoding.h +++ b/src/parquet/encodings/delta-bit-pack-encoding.h @@ -33,8 +33,8 @@ class DeltaBitPackDecoder : public Decoder<DType> { public: typedef typename DType::c_type T; - explicit DeltaBitPackDecoder(const ColumnDescriptor* descr, - MemoryAllocator* allocator = default_allocator()) + explicit DeltaBitPackDecoder( + const ColumnDescriptor* descr, MemoryAllocator* allocator = default_allocator()) : Decoder<DType>(descr, Encoding::DELTA_BINARY_PACKED), delta_bit_widths_(0, allocator) { if (DType::type_num != Type::INT32 && DType::type_num != Type::INT64) { @@ -60,9 +60,7 @@ class DeltaBitPackDecoder : public Decoder<DType> { int32_t block_size; if (!decoder_.GetVlqInt(&block_size)) ParquetException::EofException(); if (!decoder_.GetVlqInt(&num_mini_blocks_)) ParquetException::EofException(); - if (!decoder_.GetVlqInt(&values_current_block_)) { - ParquetException::EofException(); - } + if (!decoder_.GetVlqInt(&values_current_block_)) { ParquetException::EofException(); } if (!decoder_.GetZigZagVlqInt(&last_value_)) ParquetException::EofException(); delta_bit_widths_.Resize(num_mini_blocks_); @@ -119,6 +117,6 @@ class DeltaBitPackDecoder : public Decoder<DType> { int32_t last_value_; }; -} // namespace parquet +} // namespace parquet #endif http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/encodings/delta-byte-array-encoding.h ---------------------------------------------------------------------- diff --git a/src/parquet/encodings/delta-byte-array-encoding.h b/src/parquet/encodings/delta-byte-array-encoding.h index 34867e2..27d6065 100644 --- a/src/parquet/encodings/delta-byte-array-encoding.h +++ b/src/parquet/encodings/delta-byte-array-encoding.h @@ -28,12 +28,11 @@ namespace parquet { class DeltaByteArrayDecoder : public Decoder<ByteArrayType> { public: - explicit DeltaByteArrayDecoder(const ColumnDescriptor* descr, - MemoryAllocator* allocator = default_allocator()) + explicit DeltaByteArrayDecoder( + const ColumnDescriptor* descr, MemoryAllocator* allocator = default_allocator()) : Decoder<ByteArrayType>(descr, Encoding::DELTA_BYTE_ARRAY), - prefix_len_decoder_(nullptr, allocator), - suffix_decoder_(nullptr, allocator) { - } + prefix_len_decoder_(nullptr, allocator), + suffix_decoder_(nullptr, allocator) {} virtual void SetData(int num_values, const uint8_t* data, int len) { num_values_ = num_values; @@ -51,7 +50,7 @@ class DeltaByteArrayDecoder : public Decoder<ByteArrayType> { // new strings to store the results. virtual int Decode(ByteArray* buffer, int max_values) { max_values = std::min(max_values, num_values_); - for (int i = 0; i < max_values; ++i) { + for (int i = 0; i < max_values; ++i) { int prefix_len = 0; prefix_len_decoder_.Decode(&prefix_len, 1); ByteArray suffix; @@ -77,6 +76,6 @@ class DeltaByteArrayDecoder : public Decoder<ByteArrayType> { ByteArray last_value_; }; -} // namespace parquet +} // namespace parquet #endif http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/encodings/delta-length-byte-array-encoding.h ---------------------------------------------------------------------- diff --git a/src/parquet/encodings/delta-length-byte-array-encoding.h b/src/parquet/encodings/delta-length-byte-array-encoding.h index 7a19aa3..b159171 100644 --- a/src/parquet/encodings/delta-length-byte-array-encoding.h +++ b/src/parquet/encodings/delta-length-byte-array-encoding.h @@ -29,11 +29,10 @@ namespace parquet { class DeltaLengthByteArrayDecoder : public Decoder<ByteArrayType> { public: - explicit DeltaLengthByteArrayDecoder(const ColumnDescriptor* descr, - MemoryAllocator* allocator = default_allocator()) : - Decoder<ByteArrayType>(descr, Encoding::DELTA_LENGTH_BYTE_ARRAY), - len_decoder_(nullptr, allocator) { - } + explicit DeltaLengthByteArrayDecoder( + const ColumnDescriptor* descr, MemoryAllocator* allocator = default_allocator()) + : Decoder<ByteArrayType>(descr, Encoding::DELTA_LENGTH_BYTE_ARRAY), + len_decoder_(nullptr, allocator) {} virtual void SetData(int num_values, const uint8_t* data, int len) { num_values_ = num_values; @@ -49,7 +48,7 @@ class DeltaLengthByteArrayDecoder : public Decoder<ByteArrayType> { max_values = std::min(max_values, num_values_); int lengths[max_values]; len_decoder_.Decode(lengths, max_values); - for (int i = 0; i < max_values; ++i) { + for (int i = 0; i < max_values; ++i) { buffer[i].len = lengths[i]; buffer[i].ptr = data_; data_ += lengths[i]; @@ -66,6 +65,6 @@ class DeltaLengthByteArrayDecoder : public Decoder<ByteArrayType> { int len_; }; -} // namespace parquet +} // namespace parquet #endif http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/encodings/dictionary-encoding.h ---------------------------------------------------------------------- diff --git a/src/parquet/encodings/dictionary-encoding.h b/src/parquet/encodings/dictionary-encoding.h index e26ba2d..7d6785e 100644 --- a/src/parquet/encodings/dictionary-encoding.h +++ b/src/parquet/encodings/dictionary-encoding.h @@ -44,10 +44,11 @@ class DictionaryDecoder : public Decoder<Type> { // Initializes the dictionary with values from 'dictionary'. The data in // dictionary is not guaranteed to persist in memory after this call so the // dictionary decoder needs to copy the data out if necessary. - explicit DictionaryDecoder(const ColumnDescriptor* descr, - MemoryAllocator* allocator = default_allocator()): - Decoder<Type>(descr, Encoding::RLE_DICTIONARY), dictionary_(0, allocator), - byte_array_data_(0, allocator) {} + explicit DictionaryDecoder( + const ColumnDescriptor* descr, MemoryAllocator* allocator = default_allocator()) + : Decoder<Type>(descr, Encoding::RLE_DICTIONARY), + dictionary_(0, allocator), + byte_array_data_(0, allocator) {} // Perform type-specific initiatialization void SetDict(Decoder<Type>* dictionary); @@ -97,8 +98,7 @@ inline void DictionaryDecoder<Type>::SetDict(Decoder<Type>* dictionary) { } template <> -inline void DictionaryDecoder<BooleanType>::SetDict( - Decoder<BooleanType>* dictionary) { +inline void DictionaryDecoder<BooleanType>::SetDict(Decoder<BooleanType>* dictionary) { ParquetException::NYI("Dictionary encoding is not implemented for boolean values"); } @@ -129,7 +129,7 @@ inline void DictionaryDecoder<FLBAType>::SetDict(Decoder<FLBAType>* dictionary) dictionary->Decode(&dictionary_[0], num_dictionary_values); int fixed_len = descr_->type_length(); - int total_size = num_dictionary_values*fixed_len; + int total_size = num_dictionary_values * fixed_len; byte_array_data_.Resize(total_size); int offset = 0; @@ -162,9 +162,7 @@ static constexpr double MAX_HASH_LOAD = 0.7; /// the encoder, including new dictionary entries. class DictEncoderBase { public: - virtual ~DictEncoderBase() { - DCHECK(buffered_indices_.empty()); - } + virtual ~DictEncoderBase() { DCHECK(buffered_indices_.empty()); } /// Writes out the encoded dictionary to buffer. buffer must be preallocated to /// dict_encoded_size() bytes. @@ -200,17 +198,15 @@ class DictEncoderBase { int dict_encoded_size() { return dict_encoded_size_; } protected: - explicit DictEncoderBase(MemPool* pool, MemoryAllocator* allocator) : - hash_table_size_(INITIAL_HASH_TABLE_SIZE), - mod_bitmask_(hash_table_size_ - 1), - hash_slots_(0, allocator), - allocator_(allocator), - pool_(pool), - dict_encoded_size_(0) { + explicit DictEncoderBase(MemPool* pool, MemoryAllocator* allocator) + : hash_table_size_(INITIAL_HASH_TABLE_SIZE), + mod_bitmask_(hash_table_size_ - 1), + hash_slots_(0, allocator), + allocator_(allocator), + pool_(pool), + dict_encoded_size_(0) { hash_slots_.Assign(hash_table_size_, HASH_SLOT_EMPTY); - if (!CpuInfo::initialized()) { - CpuInfo::Init(); - } + if (!CpuInfo::initialized()) { CpuInfo::Init(); } } /// Size of the table. Must be a power of 2. @@ -240,18 +236,14 @@ template <typename T> class DictEncoder : public DictEncoderBase { public: explicit DictEncoder(MemPool* pool = nullptr, - MemoryAllocator* allocator = default_allocator(), int type_length = -1) : - DictEncoderBase(pool, allocator), type_length_(type_length) {} + MemoryAllocator* allocator = default_allocator(), int type_length = -1) + : DictEncoderBase(pool, allocator), type_length_(type_length) {} // TODO(wesm): think about how to address the construction semantics in // encodings/dictionary-encoding.h - void set_mem_pool(MemPool* pool) { - pool_ = pool; - } + void set_mem_pool(MemPool* pool) { pool_ = pool; } - void set_type_length(int type_length) { - type_length_ = type_length; - } + void set_type_length(int type_length) { type_length_ = type_length; } /// Encode value. Note that this does not actually write any data, just /// buffers the value's index to be written later. @@ -278,19 +270,18 @@ class DictEncoder : public DictEncoderBase { void AddDictKey(const T& value); }; -template<typename T> +template <typename T> inline int DictEncoder<T>::Hash(const T& value) const { return HashUtil::Hash(&value, sizeof(value), 0); } -template<> +template <> inline int DictEncoder<ByteArray>::Hash(const ByteArray& value) const { return HashUtil::Hash(value.ptr, value.len, 0); } -template<> -inline int DictEncoder<FixedLenByteArray>::Hash( - const FixedLenByteArray& value) const { +template <> +inline int DictEncoder<FixedLenByteArray>::Hash(const FixedLenByteArray& value) const { return HashUtil::Hash(value.ptr, type_length_, 0); } @@ -324,8 +315,7 @@ inline void DictEncoder<T>::Put(const T& v) { hash_slots_[j] = index; AddDictKey(v); - if (UNLIKELY(static_cast<int>(uniques_.size()) > - hash_table_size_ * MAX_HASH_LOAD)) { + if (UNLIKELY(static_cast<int>(uniques_.size()) > hash_table_size_ * MAX_HASH_LOAD)) { DoubleTableSize(); } } @@ -343,9 +333,7 @@ inline void DictEncoder<T>::DoubleTableSize() { for (int i = 0; i < hash_table_size_; ++i) { index = hash_slots_[i]; - if (index == HASH_SLOT_EMPTY) { - continue; - } + if (index == HASH_SLOT_EMPTY) { continue; } // Compute the hash value mod the new table size to start looking for an // empty slot @@ -370,24 +358,22 @@ inline void DictEncoder<T>::DoubleTableSize() { hash_slots_.Swap(new_hash_slots); } -template<typename T> +template <typename T> inline void DictEncoder<T>::AddDictKey(const T& v) { uniques_.push_back(v); dict_encoded_size_ += sizeof(T); } -template<> +template <> inline void DictEncoder<ByteArray>::AddDictKey(const ByteArray& v) { uint8_t* heap = pool_->Allocate(v.len); - if (UNLIKELY(v.len > 0 && heap == nullptr)) { - throw ParquetException("out of memory"); - } + if (UNLIKELY(v.len > 0 && heap == nullptr)) { throw ParquetException("out of memory"); } memcpy(heap, v.ptr, v.len); uniques_.push_back(ByteArray(v.len, heap)); dict_encoded_size_ += v.len + sizeof(uint32_t); } -template<> +template <> inline void DictEncoder<FixedLenByteArray>::AddDictKey(const FixedLenByteArray& v) { uint8_t* heap = pool_->Allocate(type_length_); if (UNLIKELY(type_length_ > 0 && heap == nullptr)) { @@ -440,6 +426,6 @@ inline int DictEncoderBase::WriteIndices(uint8_t* buffer, int buffer_len) { return 1 + encoder.len(); } -} // namespace parquet +} // namespace parquet #endif http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/encodings/encoder.h ---------------------------------------------------------------------- diff --git a/src/parquet/encodings/encoder.h b/src/parquet/encodings/encoder.h index 0d69111..8555930 100644 --- a/src/parquet/encodings/encoder.h +++ b/src/parquet/encodings/encoder.h @@ -44,8 +44,8 @@ class Encoder { const Encoding::type encoding() const { return encoding_; } protected: - explicit Encoder(const ColumnDescriptor* descr, - const Encoding::type& encoding, MemoryAllocator* allocator) + explicit Encoder(const ColumnDescriptor* descr, const Encoding::type& encoding, + MemoryAllocator* allocator) : descr_(descr), encoding_(encoding), allocator_(allocator) {} // For accessing type-specific metadata, like FIXED_LEN_BYTE_ARRAY @@ -54,6 +54,6 @@ class Encoder { MemoryAllocator* allocator_; }; -} // namespace parquet +} // namespace parquet -#endif // PARQUET_ENCODINGS_ENCODER_H +#endif // PARQUET_ENCODINGS_ENCODER_H http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/encodings/encoding-test.cc ---------------------------------------------------------------------- diff --git a/src/parquet/encodings/encoding-test.cc b/src/parquet/encodings/encoding-test.cc index d55de76..bde6f10 100644 --- a/src/parquet/encodings/encoding-test.cc +++ b/src/parquet/encodings/encoding-test.cc @@ -74,8 +74,8 @@ TEST(VectorBooleanTest, TestEncodeDecode) { template <typename T> void GenerateData(int num_values, T* out, vector<uint8_t>* heap) { // seed the prng so failure is deterministic - random_numbers(num_values, 0, std::numeric_limits<T>::min(), - std::numeric_limits<T>::max(), out); + random_numbers( + num_values, 0, std::numeric_limits<T>::min(), std::numeric_limits<T>::max(), out); } template <> @@ -148,15 +148,11 @@ class TestEncodingBase : public ::testing::Test { void SetUp() { descr_ = ExampleDescr<T>(); - if (descr_) { - type_length_ = descr_->type_length(); - } + if (descr_) { type_length_ = descr_->type_length(); } allocator_ = default_allocator(); } - void TearDown() { - pool_.FreeAll(); - } + void TearDown() { pool_.FreeAll(); } void InitData(int nvalues, int repeats) { num_values_ = nvalues * repeats; @@ -210,7 +206,6 @@ class TestEncodingBase : public ::testing::Test { using TestEncodingBase<Type>::encode_buffer_; \ using TestEncodingBase<Type>::decode_buf_; - template <typename Type> class TestPlainEncoding : public TestEncodingBase<Type> { public: @@ -225,8 +220,7 @@ class TestPlainEncoding : public TestEncodingBase<Type> { encode_buffer_ = dst.GetBuffer(); - decoder.SetData(num_values_, encode_buffer_->data(), - encode_buffer_->size()); + decoder.SetData(num_values_, encode_buffer_->data(), encode_buffer_->size()); int values_decoded = decoder.Decode(decode_buf_, num_values_); ASSERT_EQ(num_values_, values_decoded); VerifyResults<T>(decode_buf_, draws_, num_values_); @@ -246,7 +240,7 @@ TYPED_TEST(TestPlainEncoding, BasicRoundTrip) { // Dictionary encoding tests typedef ::testing::Types<Int32Type, Int64Type, Int96Type, FloatType, DoubleType, - ByteArrayType, FLBAType> DictEncodedTypes; + ByteArrayType, FLBAType> DictEncodedTypes; template <typename Type> class TestDictionaryEncoding : public TestEncodingBase<Type> { @@ -260,23 +254,21 @@ class TestDictionaryEncoding : public TestEncodingBase<Type> { dict_buffer_ = std::make_shared<OwnedMutableBuffer>(); auto indices = std::make_shared<OwnedMutableBuffer>(); - ASSERT_NO_THROW( - { - for (int i = 0; i < num_values_; ++i) { - encoder.Put(draws_[i]); - } - }); + ASSERT_NO_THROW({ + for (int i = 0; i < num_values_; ++i) { + encoder.Put(draws_[i]); + } + }); dict_buffer_->Resize(encoder.dict_encoded_size()); encoder.WriteDict(dict_buffer_->mutable_data()); indices->Resize(encoder.EstimatedDataEncodedSize()); - int actual_bytes = encoder.WriteIndices(indices->mutable_data(), - indices->size()); + int actual_bytes = encoder.WriteIndices(indices->mutable_data(), indices->size()); indices->Resize(actual_bytes); PlainDecoder<Type> dict_decoder(descr_.get()); - dict_decoder.SetData(encoder.num_entries(), dict_buffer_->data(), - dict_buffer_->size()); + dict_decoder.SetData( + encoder.num_entries(), dict_buffer_->data(), dict_buffer_->size()); DictionaryDecoder<Type> decoder(descr_.get()); decoder.SetDict(&dict_decoder); @@ -309,6 +301,6 @@ TEST(TestDictionaryEncoding, CannotDictDecodeBoolean) { ASSERT_THROW(decoder.SetDict(&dict_decoder), ParquetException); } -} // namespace test +} // namespace test -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/encodings/plain-encoding.h ---------------------------------------------------------------------- diff --git a/src/parquet/encodings/plain-encoding.h b/src/parquet/encodings/plain-encoding.h index 56243c8..71ae740 100644 --- a/src/parquet/encodings/plain-encoding.h +++ b/src/parquet/encodings/plain-encoding.h @@ -39,9 +39,8 @@ class PlainDecoder : public Decoder<DType> { typedef typename DType::c_type T; using Decoder<DType>::num_values_; - explicit PlainDecoder(const ColumnDescriptor* descr) : - Decoder<DType>(descr, Encoding::PLAIN), - data_(NULL), len_(0) { + explicit PlainDecoder(const ColumnDescriptor* descr) + : Decoder<DType>(descr, Encoding::PLAIN), data_(NULL), len_(0) { if (descr_ && descr_->physical_type() == Type::FIXED_LEN_BYTE_ARRAY) { type_length_ = descr_->type_length(); } else { @@ -66,12 +65,10 @@ class PlainDecoder : public Decoder<DType> { // Decode routine templated on C++ type rather than type enum template <typename T> -inline int DecodePlain(const uint8_t* data, int64_t data_size, int num_values, - int type_length, T* out) { +inline int DecodePlain( + const uint8_t* data, int64_t data_size, int num_values, int type_length, T* out) { int bytes_to_decode = num_values * sizeof(T); - if (data_size < bytes_to_decode) { - ParquetException::EofException(); - } + if (data_size < bytes_to_decode) { ParquetException::EofException(); } memcpy(out, data, bytes_to_decode); return bytes_to_decode; } @@ -101,9 +98,7 @@ template <> inline int DecodePlain<FixedLenByteArray>(const uint8_t* data, int64_t data_size, int num_values, int type_length, FixedLenByteArray* out) { int bytes_to_decode = type_length * num_values; - if (data_size < bytes_to_decode) { - ParquetException::EofException(); - } + if (data_size < bytes_to_decode) { ParquetException::EofException(); } for (int i = 0; i < num_values; ++i) { out[i].ptr = data; data += type_length; @@ -115,8 +110,7 @@ inline int DecodePlain<FixedLenByteArray>(const uint8_t* data, int64_t data_size template <typename DType> inline int PlainDecoder<DType>::Decode(T* buffer, int max_values) { max_values = std::min(max_values, num_values_); - int bytes_consumed = DecodePlain<T>(data_, len_, max_values, - type_length_, buffer); + int bytes_consumed = DecodePlain<T>(data_, len_, max_values, type_length_, buffer); data_ += bytes_consumed; len_ -= bytes_consumed; num_values_ -= max_values; @@ -126,8 +120,8 @@ inline int PlainDecoder<DType>::Decode(T* buffer, int max_values) { template <> class PlainDecoder<BooleanType> : public Decoder<BooleanType> { public: - explicit PlainDecoder(const ColumnDescriptor* descr) : - Decoder<BooleanType>(descr, Encoding::PLAIN) {} + explicit PlainDecoder(const ColumnDescriptor* descr) + : Decoder<BooleanType>(descr, Encoding::PLAIN) {} virtual void SetData(int num_values, const uint8_t* data, int len) { num_values_ = num_values; @@ -139,9 +133,7 @@ class PlainDecoder<BooleanType> : public Decoder<BooleanType> { max_values = std::min(max_values, num_values_); bool val; for (int i = 0; i < max_values; ++i) { - if (!bit_reader_.GetValue(1, &val)) { - ParquetException::EofException(); - } + if (!bit_reader_.GetValue(1, &val)) { ParquetException::EofException(); } BitUtil::SetArrayBit(buffer, i, val); } num_values_ -= max_values; @@ -152,9 +144,7 @@ class PlainDecoder<BooleanType> : public Decoder<BooleanType> { max_values = std::min(max_values, num_values_); bool val; for (int i = 0; i < max_values; ++i) { - if (!bit_reader_.GetValue(1, &val)) { - ParquetException::EofException(); - } + if (!bit_reader_.GetValue(1, &val)) { ParquetException::EofException(); } buffer[i] = val; } num_values_ -= max_values; @@ -173,9 +163,9 @@ class PlainEncoder : public Encoder<DType> { public: typedef typename DType::c_type T; - explicit PlainEncoder(const ColumnDescriptor* descr, - MemoryAllocator* allocator = default_allocator()) : - Encoder<DType>(descr, Encoding::PLAIN, allocator) {} + explicit PlainEncoder( + const ColumnDescriptor* descr, MemoryAllocator* allocator = default_allocator()) + : Encoder<DType>(descr, Encoding::PLAIN, allocator) {} void Encode(const T* src, int num_values, OutputStream* dst) override; }; @@ -183,9 +173,9 @@ class PlainEncoder : public Encoder<DType> { template <> class PlainEncoder<BooleanType> : public Encoder<BooleanType> { public: - explicit PlainEncoder(const ColumnDescriptor* descr, - MemoryAllocator* allocator = default_allocator()) : - Encoder<BooleanType>(descr, Encoding::PLAIN, allocator) {} + explicit PlainEncoder( + const ColumnDescriptor* descr, MemoryAllocator* allocator = default_allocator()) + : Encoder<BooleanType>(descr, Encoding::PLAIN, allocator) {} virtual void Encode(const bool* src, int num_values, OutputStream* dst) { int bytes_required = BitUtil::Ceil(num_values, 8); @@ -222,14 +212,14 @@ class PlainEncoder<BooleanType> : public Encoder<BooleanType> { }; template <typename DType> -inline void PlainEncoder<DType>::Encode(const T* buffer, int num_values, - OutputStream* dst) { +inline void PlainEncoder<DType>::Encode( + const T* buffer, int num_values, OutputStream* dst) { dst->Write(reinterpret_cast<const uint8_t*>(buffer), num_values * sizeof(T)); } template <> -inline void PlainEncoder<ByteArrayType>::Encode(const ByteArray* src, - int num_values, OutputStream* dst) { +inline void PlainEncoder<ByteArrayType>::Encode( + const ByteArray* src, int num_values, OutputStream* dst) { for (int i = 0; i < num_values; ++i) { // Write the result to the output stream dst->Write(reinterpret_cast<const uint8_t*>(&src[i].len), sizeof(uint32_t)); @@ -245,6 +235,6 @@ inline void PlainEncoder<FLBAType>::Encode( dst->Write(reinterpret_cast<const uint8_t*>(src[i].ptr), descr_->type_length()); } } -} // namespace parquet +} // namespace parquet #endif http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/exception.h ---------------------------------------------------------------------- diff --git a/src/parquet/exception.h b/src/parquet/exception.h index 608a45a..03f2356 100644 --- a/src/parquet/exception.h +++ b/src/parquet/exception.h @@ -44,6 +44,6 @@ class ParquetException : public std::exception { std::string msg_; }; -} // namespace parquet +} // namespace parquet -#endif // PARQUET_EXCEPTION_H +#endif // PARQUET_EXCEPTION_H http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/file/file-deserialize-test.cc ---------------------------------------------------------------------- diff --git a/src/parquet/file/file-deserialize-test.cc b/src/parquet/file/file-deserialize-test.cc index be9c23c..db99f16 100644 --- a/src/parquet/file/file-deserialize-test.cc +++ b/src/parquet/file/file-deserialize-test.cc @@ -39,17 +39,14 @@ namespace parquet { - // Adds page statistics occupying a certain amount of bytes (for testing very // large page headers) -static inline void AddDummyStats(int stat_size, - format::DataPageHeader& data_page) { - +static inline void AddDummyStats(int stat_size, format::DataPageHeader& data_page) { std::vector<uint8_t> stat_bytes(stat_size); // Some non-zero value std::fill(stat_bytes.begin(), stat_bytes.end(), 1); - data_page.statistics.__set_max(std::string( - reinterpret_cast<const char*>(stat_bytes.data()), stat_size)); + data_page.statistics.__set_max( + std::string(reinterpret_cast<const char*>(stat_bytes.data()), stat_size)); data_page.__isset.statistics = true; } @@ -63,16 +60,15 @@ class TestPageSerde : public ::testing::Test { ResetStream(); } - void InitSerializedPageReader(Compression::type codec = - Compression::UNCOMPRESSED) { + void InitSerializedPageReader(Compression::type codec = Compression::UNCOMPRESSED) { EndStream(); std::unique_ptr<InputStream> stream; stream.reset(new InMemoryInputStream(out_buffer_)); page_reader_.reset(new SerializedPageReader(std::move(stream), codec)); } - void WriteDataPageHeader(int max_serialized_len = 1024, - int32_t uncompressed_size = 0, int32_t compressed_size = 0) { + void WriteDataPageHeader(int max_serialized_len = 1024, int32_t uncompressed_size = 0, + int32_t compressed_size = 0) { // Simplifying writing serialized data page headers which may or may not // have meaningful data associated with them @@ -82,17 +78,13 @@ class TestPageSerde : public ::testing::Test { page_header_.compressed_page_size = compressed_size; page_header_.type = format::PageType::DATA_PAGE; - ASSERT_NO_THROW(SerializeThriftMsg(&page_header_, max_serialized_len, - out_stream_.get())); + ASSERT_NO_THROW( + SerializeThriftMsg(&page_header_, max_serialized_len, out_stream_.get())); } - void ResetStream() { - out_stream_.reset(new InMemoryOutputStream); - } + void ResetStream() { out_stream_.reset(new InMemoryOutputStream); } - void EndStream() { - out_buffer_ = out_stream_->GetBuffer(); - } + void EndStream() { out_buffer_ = out_stream_->GetBuffer(); } protected: std::unique_ptr<InMemoryOutputStream> out_stream_; @@ -103,25 +95,22 @@ class TestPageSerde : public ::testing::Test { format::DataPageHeader data_page_header_; }; -void CheckDataPageHeader(const format::DataPageHeader expected, - const Page* page) { +void CheckDataPageHeader(const format::DataPageHeader expected, const Page* page) { ASSERT_EQ(PageType::DATA_PAGE, page->type()); const DataPage* data_page = static_cast<const DataPage*>(page); ASSERT_EQ(expected.num_values, data_page->num_values()); ASSERT_EQ(expected.encoding, data_page->encoding()); - ASSERT_EQ(expected.definition_level_encoding, - data_page->definition_level_encoding()); - ASSERT_EQ(expected.repetition_level_encoding, - data_page->repetition_level_encoding()); + ASSERT_EQ(expected.definition_level_encoding, data_page->definition_level_encoding()); + ASSERT_EQ(expected.repetition_level_encoding, data_page->repetition_level_encoding()); if (expected.statistics.__isset.max) { - ASSERT_EQ(0, memcmp(expected.statistics.max.c_str(), - data_page->max(), expected.statistics.max.length())); + ASSERT_EQ(0, memcmp(expected.statistics.max.c_str(), data_page->max(), + expected.statistics.max.length())); } if (expected.statistics.__isset.min) { - ASSERT_EQ(0, memcmp(expected.statistics.min.c_str(), - data_page->min(), expected.statistics.min.length())); + ASSERT_EQ(0, memcmp(expected.statistics.min.c_str(), data_page->min(), + expected.statistics.min.length())); } } @@ -139,13 +128,13 @@ TEST_F(TestPageSerde, DataPage) { } TEST_F(TestPageSerde, TestLargePageHeaders) { - int stats_size = 256 * 1024; // 256 KB + int stats_size = 256 * 1024; // 256 KB AddDummyStats(stats_size, data_page_header_); // Any number to verify metadata roundtrip data_page_header_.num_values = 4141; - int max_header_size = 512 * 1024; // 512 KB + int max_header_size = 512 * 1024; // 512 KB WriteDataPageHeader(max_header_size); ASSERT_GE(max_header_size, out_stream_->Tell()); @@ -159,11 +148,11 @@ TEST_F(TestPageSerde, TestLargePageHeaders) { } TEST_F(TestPageSerde, TestFailLargePageHeaders) { - int stats_size = 256 * 1024; // 256 KB + int stats_size = 256 * 1024; // 256 KB AddDummyStats(stats_size, data_page_header_); // Serialize the Page header - int max_header_size = 512 * 1024; // 512 KB + int max_header_size = 512 * 1024; // 512 KB WriteDataPageHeader(max_header_size); ASSERT_GE(max_header_size, out_stream_->Tell()); @@ -185,7 +174,7 @@ TEST_F(TestPageSerde, Compression) { int num_pages = 10; - std::vector<std::vector<uint8_t> > faux_data; + std::vector<std::vector<uint8_t>> faux_data; faux_data.resize(num_pages); for (int i = 0; i < num_pages; ++i) { // The pages keep getting larger @@ -203,8 +192,8 @@ TEST_F(TestPageSerde, Compression) { int64_t max_compressed_size = codec->MaxCompressedLen(data_size, data); buffer.resize(max_compressed_size); - int64_t actual_size = codec->Compress(data_size, data, - max_compressed_size, &buffer[0]); + int64_t actual_size = + codec->Compress(data_size, data, max_compressed_size, &buffer[0]); WriteDataPageHeader(1024, data_size, actual_size); out_stream_->Write(buffer.data(), actual_size); @@ -245,8 +234,7 @@ class TestParquetFileReader : public ::testing::Test { reader_.reset(new ParquetFileReader()); ASSERT_THROW( - reader_->Open(SerializedFile::Open(std::move(reader))), - ParquetException); + reader_->Open(SerializedFile::Open(std::move(reader))), ParquetException); } protected: @@ -291,4 +279,4 @@ TEST_F(TestParquetFileReader, IncompleteMetadata) { AssertInvalidFileThrows(buffer); } -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/file/file-serialize-test.cc ---------------------------------------------------------------------- diff --git a/src/parquet/file/file-serialize-test.cc b/src/parquet/file/file-serialize-test.cc index a75d250..194e496 100644 --- a/src/parquet/file/file-serialize-test.cc +++ b/src/parquet/file/file-serialize-test.cc @@ -37,35 +37,32 @@ class TestSerialize : public ::testing::Test { public: void SetUpSchemaRequired() { auto pnode = PrimitiveNode::Make("int64", Repetition::REQUIRED, Type::INT64); - node_ = GroupNode::Make("schema", Repetition::REQUIRED, - std::vector<NodePtr>({pnode})); + node_ = + GroupNode::Make("schema", Repetition::REQUIRED, std::vector<NodePtr>({pnode})); schema_.Init(node_); } void SetUpSchemaOptional() { auto pnode = PrimitiveNode::Make("int64", Repetition::OPTIONAL, Type::INT64); - node_ = GroupNode::Make("schema", Repetition::REQUIRED, - std::vector<NodePtr>({pnode})); + node_ = + GroupNode::Make("schema", Repetition::REQUIRED, std::vector<NodePtr>({pnode})); schema_.Init(node_); } void SetUpSchemaRepeated() { auto pnode = PrimitiveNode::Make("int64", Repetition::REPEATED, Type::INT64); - node_ = GroupNode::Make("schema", Repetition::REQUIRED, - std::vector<NodePtr>({pnode})); + node_ = + GroupNode::Make("schema", Repetition::REQUIRED, std::vector<NodePtr>({pnode})); schema_.Init(node_); } - void SetUp() { - SetUpSchemaRequired(); - } + void SetUp() { SetUpSchemaRequired(); } protected: NodePtr node_; SchemaDescriptor schema_; }; - TEST_F(TestSerialize, SmallFile) { std::shared_ptr<InMemoryOutputStream> sink(new InMemoryOutputStream()); auto gnode = std::static_pointer_cast<GroupNode>(node_); @@ -100,6 +97,6 @@ TEST_F(TestSerialize, SmallFile) { ASSERT_EQ(values, values_out); } -} // namespace test +} // namespace test -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/file/reader-internal.cc ---------------------------------------------------------------------- diff --git a/src/parquet/file/reader-internal.cc b/src/parquet/file/reader-internal.cc index 20e089a..905c357 100644 --- a/src/parquet/file/reader-internal.cc +++ b/src/parquet/file/reader-internal.cc @@ -42,9 +42,8 @@ namespace parquet { // assembled in a serialized stream for storing in a Parquet files SerializedPageReader::SerializedPageReader(std::unique_ptr<InputStream> stream, - Compression::type codec_type, MemoryAllocator* allocator) : - stream_(std::move(stream)), - decompression_buffer_(0, allocator) { + Compression::type codec_type, MemoryAllocator* allocator) + : stream_(std::move(stream)), decompression_buffer_(0, allocator) { max_page_header_size_ = DEFAULT_MAX_PAGE_HEADER_SIZE; decompressor_ = Codec::Create(codec_type); } @@ -65,9 +64,7 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() { // until a maximum allowed header limit while (true) { buffer = stream_->Peek(allowed_page_size, &bytes_available); - if (bytes_available == 0) { - return std::shared_ptr<Page>(nullptr); - } + if (bytes_available == 0) { return std::shared_ptr<Page>(nullptr); } // This gets used, then set by DeserializeThriftMsg header_size = bytes_available; @@ -100,8 +97,8 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() { if (uncompressed_len > static_cast<int>(decompression_buffer_.size())) { decompression_buffer_.Resize(uncompressed_len); } - decompressor_->Decompress(compressed_len, buffer, uncompressed_len, - &decompression_buffer_[0]); + decompressor_->Decompress( + compressed_len, buffer, uncompressed_len, &decompression_buffer_[0]); buffer = &decompression_buffer_[0]; } @@ -109,40 +106,32 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() { if (current_page_header_.type == format::PageType::DICTIONARY_PAGE) { const format::DictionaryPageHeader& dict_header = - current_page_header_.dictionary_page_header; + current_page_header_.dictionary_page_header; - bool is_sorted = dict_header.__isset.is_sorted? dict_header.is_sorted : false; + bool is_sorted = dict_header.__isset.is_sorted ? dict_header.is_sorted : false; - return std::make_shared<DictionaryPage>(page_buffer, - dict_header.num_values, FromThrift(dict_header.encoding), - is_sorted); + return std::make_shared<DictionaryPage>(page_buffer, dict_header.num_values, + FromThrift(dict_header.encoding), is_sorted); } else if (current_page_header_.type == format::PageType::DATA_PAGE) { const format::DataPageHeader& header = current_page_header_.data_page_header; - auto page = std::make_shared<DataPage>(page_buffer, - header.num_values, - FromThrift(header.encoding), - FromThrift(header.definition_level_encoding), + auto page = std::make_shared<DataPage>(page_buffer, header.num_values, + FromThrift(header.encoding), FromThrift(header.definition_level_encoding), FromThrift(header.repetition_level_encoding)); if (header.__isset.statistics) { const format::Statistics stats = header.statistics; - if (stats.__isset.max) { - page->max_ = stats.max; - } - if (stats.__isset.min) { - page->min_ = stats.min; - } + if (stats.__isset.max) { page->max_ = stats.max; } + if (stats.__isset.min) { page->min_ = stats.min; } } return page; } else if (current_page_header_.type == format::PageType::DATA_PAGE_V2) { const format::DataPageHeaderV2& header = current_page_header_.data_page_header_v2; - bool is_compressed = header.__isset.is_compressed? header.is_compressed : false; - return std::make_shared<DataPageV2>(page_buffer, - header.num_values, header.num_nulls, header.num_rows, - FromThrift(header.encoding), - header.definition_levels_byte_length, - header.repetition_levels_byte_length, is_compressed); + bool is_compressed = header.__isset.is_compressed ? header.is_compressed : false; + return std::make_shared<DataPageV2>(page_buffer, header.num_values, + header.num_nulls, header.num_rows, FromThrift(header.encoding), + header.definition_levels_byte_length, header.repetition_levels_byte_length, + is_compressed); } else { // We don't know what this page type is. We're allowed to skip non-data // pages. @@ -181,8 +170,8 @@ std::unique_ptr<PageReader> SerializedRowGroup::GetColumnPageReader(int i) { } std::unique_ptr<InputStream> stream(new InMemoryInputStream(buffer)); - return std::unique_ptr<PageReader>(new SerializedPageReader(std::move(stream), - FromThrift(col.meta_data.codec), allocator_)); + return std::unique_ptr<PageReader>(new SerializedPageReader( + std::move(stream), FromThrift(col.meta_data.codec), allocator_)); } RowGroupStatistics SerializedRowGroup::GetColumnStats(int i) { @@ -227,8 +216,8 @@ SerializedFile::~SerializedFile() { } std::shared_ptr<RowGroupReader> SerializedFile::GetRowGroup(int i) { - std::unique_ptr<SerializedRowGroup> contents(new SerializedRowGroup(source_.get(), - &metadata_.row_groups[i], allocator_)); + std::unique_ptr<SerializedRowGroup> contents( + new SerializedRowGroup(source_.get(), &metadata_.row_groups[i], allocator_)); return std::make_shared<RowGroupReader>(&schema_, std::move(contents), allocator_); } @@ -245,11 +234,9 @@ int SerializedFile::num_row_groups() const { return metadata_.row_groups.size(); } -SerializedFile::SerializedFile( - std::unique_ptr<RandomAccessSource> source, - MemoryAllocator* allocator = default_allocator()) : - source_(std::move(source)), allocator_(allocator) {} - +SerializedFile::SerializedFile(std::unique_ptr<RandomAccessSource> source, + MemoryAllocator* allocator = default_allocator()) + : source_(std::move(source)), allocator_(allocator) {} void SerializedFile::ParseMetaData() { int64_t filesize = source_->Size(); @@ -261,15 +248,15 @@ void SerializedFile::ParseMetaData() { uint8_t footer_buffer[FOOTER_SIZE]; source_->Seek(filesize - FOOTER_SIZE); int64_t bytes_read = source_->Read(FOOTER_SIZE, footer_buffer); - if (bytes_read != FOOTER_SIZE || - memcmp(footer_buffer + 4, PARQUET_MAGIC, 4) != 0) { + if (bytes_read != FOOTER_SIZE || memcmp(footer_buffer + 4, PARQUET_MAGIC, 4) != 0) { throw ParquetException("Invalid parquet file. Corrupt footer."); } uint32_t metadata_len = *reinterpret_cast<uint32_t*>(footer_buffer); int64_t metadata_start = filesize - FOOTER_SIZE - metadata_len; if (FOOTER_SIZE + metadata_len > filesize) { - throw ParquetException("Invalid parquet file. File is less than " + throw ParquetException( + "Invalid parquet file. File is less than " "file metadata size."); } source_->Seek(metadata_start); @@ -281,9 +268,8 @@ void SerializedFile::ParseMetaData() { } DeserializeThriftMsg(&metadata_buffer[0], &metadata_len, &metadata_); - schema::FlatSchemaConverter converter(&metadata_.schema[0], - metadata_.schema.size()); + schema::FlatSchemaConverter converter(&metadata_.schema[0], metadata_.schema.size()); schema_.Init(converter.Convert()); } -} // namespace parquet +} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/file/reader-internal.h ---------------------------------------------------------------------- diff --git a/src/parquet/file/reader-internal.h b/src/parquet/file/reader-internal.h index 1e8256a..797ff93 100644 --- a/src/parquet/file/reader-internal.h +++ b/src/parquet/file/reader-internal.h @@ -42,17 +42,15 @@ static constexpr uint32_t DEFAULT_PAGE_HEADER_SIZE = 16 * 1024; // and the page metadata. class SerializedPageReader : public PageReader { public: - SerializedPageReader(std::unique_ptr<InputStream> stream, - Compression::type codec, MemoryAllocator* allocator = default_allocator()); + SerializedPageReader(std::unique_ptr<InputStream> stream, Compression::type codec, + MemoryAllocator* allocator = default_allocator()); virtual ~SerializedPageReader() {} // Implement the PageReader interface virtual std::shared_ptr<Page> NextPage(); - void set_max_page_header_size(uint32_t size) { - max_page_header_size_ = size; - } + void set_max_page_header_size(uint32_t size) { max_page_header_size_ = size; } private: std::unique_ptr<InputStream> stream_; @@ -70,11 +68,9 @@ class SerializedPageReader : public PageReader { // RowGroupReader::Contents implementation for the Parquet file specification class SerializedRowGroup : public RowGroupReader::Contents { public: - SerializedRowGroup(RandomAccessSource* source, - const format::RowGroup* metadata, MemoryAllocator* allocator) : - source_(source), - metadata_(metadata), - allocator_(allocator) {} + SerializedRowGroup(RandomAccessSource* source, const format::RowGroup* metadata, + MemoryAllocator* allocator) + : source_(source), metadata_(metadata), allocator_(allocator) {} virtual int num_columns() const; virtual int64_t num_rows() const; @@ -108,8 +104,8 @@ class SerializedFile : public ParquetFileReader::Contents { private: // This class takes ownership of the provided data source - explicit SerializedFile(std::unique_ptr<RandomAccessSource> source, - MemoryAllocator* allocator); + explicit SerializedFile( + std::unique_ptr<RandomAccessSource> source, MemoryAllocator* allocator); std::unique_ptr<RandomAccessSource> source_; format::FileMetaData metadata_; @@ -118,6 +114,6 @@ class SerializedFile : public ParquetFileReader::Contents { void ParseMetaData(); }; -} // namespace parquet +} // namespace parquet -#endif // PARQUET_FILE_READER_INTERNAL_H +#endif // PARQUET_FILE_READER_INTERNAL_H http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/674dbb39/src/parquet/file/reader.cc ---------------------------------------------------------------------- diff --git a/src/parquet/file/reader.cc b/src/parquet/file/reader.cc index 6523711..232dbe4 100644 --- a/src/parquet/file/reader.cc +++ b/src/parquet/file/reader.cc @@ -41,10 +41,8 @@ namespace parquet { // RowGroupReader public API RowGroupReader::RowGroupReader(const SchemaDescriptor* schema, - std::unique_ptr<Contents> contents, MemoryAllocator* allocator) : - schema_(schema), - contents_(std::move(contents)), - allocator_(allocator) {} + std::unique_ptr<Contents> contents, MemoryAllocator* allocator) + : schema_(schema), contents_(std::move(contents)), allocator_(allocator) {} int RowGroupReader::num_columns() const { return contents_->num_columns(); @@ -84,8 +82,8 @@ std::unique_ptr<ParquetFileReader> ParquetFileReader::Open( return result; } -std::unique_ptr<ParquetFileReader> ParquetFileReader::OpenFile(const std::string& path, - bool memory_map, MemoryAllocator* allocator) { +std::unique_ptr<ParquetFileReader> ParquetFileReader::OpenFile( + const std::string& path, bool memory_map, MemoryAllocator* allocator) { std::unique_ptr<LocalFileSource> file; if (memory_map) { file.reset(new MemoryMapSource(allocator)); @@ -103,9 +101,7 @@ void ParquetFileReader::Open(std::unique_ptr<ParquetFileReader::Contents> conten } void ParquetFileReader::Close() { - if (contents_) { - contents_->Close(); - } + if (contents_) { contents_->Close(); } } int ParquetFileReader::num_row_groups() const { @@ -124,8 +120,7 @@ std::shared_ptr<RowGroupReader> ParquetFileReader::RowGroup(int i) { if (i >= num_row_groups()) { std::stringstream ss; ss << "The file only has " << num_row_groups() - << "row groups, requested reader for: " - << i; + << "row groups, requested reader for: " << i; throw ParquetException(ss.str()); } @@ -138,8 +133,8 @@ std::shared_ptr<RowGroupReader> ParquetFileReader::RowGroup(int i) { // the fixed initial size is just for an example #define COL_WIDTH "20" -void ParquetFileReader::DebugPrint(std::ostream& stream, - std::list<int> selected_columns, bool print_values) { +void ParquetFileReader::DebugPrint( + std::ostream& stream, std::list<int> selected_columns, bool print_values) { stream << "File statistics:\n"; stream << "Total rows: " << num_rows() << "\n"; @@ -157,11 +152,8 @@ void ParquetFileReader::DebugPrint(std::ostream& stream, for (auto i : selected_columns) { const ColumnDescriptor* descr = schema_->Column(i); - stream << "Column " << i << ": " - << descr->name() - << " (" - << type_to_string(descr->physical_type()) - << ")" << std::endl; + stream << "Column " << i << ": " << descr->name() << " (" + << type_to_string(descr->physical_type()) << ")" << std::endl; } for (int r = 0; r < num_row_groups(); ++r) { @@ -173,25 +165,19 @@ void ParquetFileReader::DebugPrint(std::ostream& stream, for (auto i : selected_columns) { RowGroupStatistics stats = group_reader->GetColumnStats(i); - stream << "Column " << i << ": " - << group_reader->num_rows() << " rows, " - << stats.num_values << " values, " - << stats.null_count << " null values, " - << stats.distinct_count << " distinct values, " - << *stats.max << " max, " - << *stats.min << " min, " - << std::endl; + stream << "Column " << i << ": " << group_reader->num_rows() << " rows, " + << stats.num_values << " values, " << stats.null_count << " null values, " + << stats.distinct_count << " distinct values, " << *stats.max << " max, " + << *stats.min << " min, " << std::endl; } - if (!print_values) { - continue; - } + if (!print_values) { continue; } static constexpr int bufsize = 25; char buffer[bufsize]; // Create readers for selected columns and print contents - vector<std::shared_ptr<Scanner> > scanners(selected_columns.size(), NULL); + vector<std::shared_ptr<Scanner>> scanners(selected_columns.size(), NULL); int j = 0; for (auto i : selected_columns) { std::shared_ptr<ColumnReader> col_reader = group_reader->Column(i); @@ -223,4 +209,4 @@ void ParquetFileReader::DebugPrint(std::ostream& stream, } } -} // namespace parquet +} // namespace parquet
