http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/84db929e/src/parquet/column/scanner.h ---------------------------------------------------------------------- diff --git a/src/parquet/column/scanner.h b/src/parquet/column/scanner.h deleted file mode 100644 index a9b83c1..0000000 --- a/src/parquet/column/scanner.h +++ /dev/null @@ -1,232 +0,0 @@ -// 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. - -#ifndef PARQUET_COLUMN_SCANNER_H -#define PARQUET_COLUMN_SCANNER_H - -#include <cstdint> -#include <memory> -#include <ostream> -#include <stdio.h> -#include <string> -#include <vector> - -#include "parquet/column/reader.h" -#include "parquet/exception.h" -#include "parquet/schema.h" -#include "parquet/types.h" -#include "parquet/util/memory.h" -#include "parquet/util/visibility.h" - -namespace parquet { - -static constexpr int64_t DEFAULT_SCANNER_BATCH_SIZE = 128; - -class PARQUET_EXPORT Scanner { - public: - explicit Scanner(std::shared_ptr<ColumnReader> reader, - int64_t batch_size = DEFAULT_SCANNER_BATCH_SIZE, - ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()) - : batch_size_(batch_size), - level_offset_(0), - levels_buffered_(0), - value_buffer_(std::make_shared<PoolBuffer>(pool)), - value_offset_(0), - values_buffered_(0), - reader_(reader) { - def_levels_.resize(descr()->max_definition_level() > 0 ? batch_size_ : 0); - rep_levels_.resize(descr()->max_repetition_level() > 0 ? batch_size_ : 0); - } - - virtual ~Scanner() {} - - static std::shared_ptr<Scanner> Make(std::shared_ptr<ColumnReader> col_reader, - int64_t batch_size = DEFAULT_SCANNER_BATCH_SIZE, - ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()); - - virtual void PrintNext(std::ostream& out, int width) = 0; - - bool HasNext() { return level_offset_ < levels_buffered_ || reader_->HasNext(); } - - const ColumnDescriptor* descr() const { return reader_->descr(); } - - int64_t batch_size() const { return batch_size_; } - - void SetBatchSize(int64_t batch_size) { batch_size_ = batch_size; } - - protected: - int64_t batch_size_; - - std::vector<int16_t> def_levels_; - std::vector<int16_t> rep_levels_; - int level_offset_; - int levels_buffered_; - - std::shared_ptr<PoolBuffer> value_buffer_; - int value_offset_; - int64_t values_buffered_; - - private: - std::shared_ptr<ColumnReader> reader_; -}; - -template <typename DType> -class PARQUET_EXPORT TypedScanner : public Scanner { - public: - typedef typename DType::c_type T; - - explicit TypedScanner(std::shared_ptr<ColumnReader> reader, - int64_t batch_size = DEFAULT_SCANNER_BATCH_SIZE, - ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()) - : Scanner(reader, batch_size, pool) { - typed_reader_ = static_cast<TypedColumnReader<DType>*>(reader.get()); - int value_byte_size = type_traits<DType::type_num>::value_byte_size; - PARQUET_THROW_NOT_OK(value_buffer_->Resize(batch_size_ * value_byte_size)); - values_ = reinterpret_cast<T*>(value_buffer_->mutable_data()); - } - - virtual ~TypedScanner() {} - - bool NextLevels(int16_t* def_level, int16_t* rep_level) { - if (level_offset_ == levels_buffered_) { - levels_buffered_ = - static_cast<int>(typed_reader_->ReadBatch(static_cast<int>(batch_size_), - def_levels_.data(), rep_levels_.data(), values_, &values_buffered_)); - - value_offset_ = 0; - level_offset_ = 0; - if (!levels_buffered_) { return false; } - } - *def_level = descr()->max_definition_level() > 0 ? def_levels_[level_offset_] : 0; - *rep_level = descr()->max_repetition_level() > 0 ? rep_levels_[level_offset_] : 0; - level_offset_++; - return true; - } - - bool Next(T* val, int16_t* def_level, int16_t* rep_level, bool* is_null) { - if (level_offset_ == levels_buffered_) { - if (!HasNext()) { - // Out of data pages - return false; - } - } - - NextLevels(def_level, rep_level); - *is_null = *def_level < descr()->max_definition_level(); - - if (*is_null) { return true; } - - if (value_offset_ == values_buffered_) { - throw ParquetException("Value was non-null, but has not been buffered"); - } - *val = values_[value_offset_++]; - return true; - } - - // Returns true if there is a next value - bool NextValue(T* val, bool* is_null) { - if (level_offset_ == levels_buffered_) { - if (!HasNext()) { - // Out of data pages - return false; - } - } - - // Out of values - int16_t def_level = -1; - int16_t rep_level = -1; - NextLevels(&def_level, &rep_level); - *is_null = def_level < descr()->max_definition_level(); - - if (*is_null) { return true; } - - if (value_offset_ == values_buffered_) { - throw ParquetException("Value was non-null, but has not been buffered"); - } - *val = values_[value_offset_++]; - return true; - } - - virtual void PrintNext(std::ostream& out, int width) { - T val; - bool is_null = false; - char buffer[25]; - - if (!NextValue(&val, &is_null)) { throw ParquetException("No more values buffered"); } - - if (is_null) { - std::string null_fmt = format_fwf<ByteArrayType>(width); - snprintf(buffer, sizeof(buffer), null_fmt.c_str(), "NULL"); - } else { - FormatValue(&val, buffer, sizeof(buffer), width); - } - out << buffer; - } - - private: - // The ownership of this object is expressed through the reader_ variable in the base - TypedColumnReader<DType>* typed_reader_; - - inline void FormatValue(void* val, char* buffer, int bufsize, int width); - - T* values_; -}; - -template <typename DType> -inline void TypedScanner<DType>::FormatValue( - void* val, char* buffer, int bufsize, int width) { - std::string fmt = format_fwf<DType>(width); - snprintf(buffer, bufsize, fmt.c_str(), *reinterpret_cast<T*>(val)); -} - -template <> -inline void TypedScanner<Int96Type>::FormatValue( - void* val, char* buffer, int bufsize, int width) { - std::string fmt = format_fwf<Int96Type>(width); - std::string result = Int96ToString(*reinterpret_cast<Int96*>(val)); - snprintf(buffer, bufsize, fmt.c_str(), result.c_str()); -} - -template <> -inline void TypedScanner<ByteArrayType>::FormatValue( - void* val, char* buffer, int bufsize, int width) { - std::string fmt = format_fwf<ByteArrayType>(width); - std::string result = ByteArrayToString(*reinterpret_cast<ByteArray*>(val)); - snprintf(buffer, bufsize, fmt.c_str(), result.c_str()); -} - -template <> -inline void TypedScanner<FLBAType>::FormatValue( - void* val, char* buffer, int bufsize, int width) { - std::string fmt = format_fwf<FLBAType>(width); - std::string result = FixedLenByteArrayToString( - *reinterpret_cast<FixedLenByteArray*>(val), descr()->type_length()); - snprintf(buffer, bufsize, fmt.c_str(), result.c_str()); -} - -typedef TypedScanner<BooleanType> BoolScanner; -typedef TypedScanner<Int32Type> Int32Scanner; -typedef TypedScanner<Int64Type> Int64Scanner; -typedef TypedScanner<Int96Type> Int96Scanner; -typedef TypedScanner<FloatType> FloatScanner; -typedef TypedScanner<DoubleType> DoubleScanner; -typedef TypedScanner<ByteArrayType> ByteArrayScanner; -typedef TypedScanner<FLBAType> FixedLenByteArrayScanner; - -} // namespace parquet - -#endif // PARQUET_COLUMN_SCANNER_H
http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/84db929e/src/parquet/column/statistics-test.cc ---------------------------------------------------------------------- diff --git a/src/parquet/column/statistics-test.cc b/src/parquet/column/statistics-test.cc deleted file mode 100644 index e656f81..0000000 --- a/src/parquet/column/statistics-test.cc +++ /dev/null @@ -1,358 +0,0 @@ -// 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 <gtest/gtest.h> - -#include <algorithm> -#include <array> -#include <cstdint> -#include <cstring> -#include <memory> -#include <vector> - -#include "parquet/column/reader.h" -#include "parquet/column/statistics.h" -#include "parquet/column/test-specialization.h" -#include "parquet/column/test-util.h" -#include "parquet/column/writer.h" -#include "parquet/file/reader.h" -#include "parquet/file/writer.h" -#include "parquet/schema.h" -#include "parquet/thrift.h" -#include "parquet/types.h" -#include "parquet/util/memory.h" - -using arrow::default_memory_pool; -using arrow::MemoryPool; - -namespace parquet { - -using schema::NodePtr; -using schema::PrimitiveNode; -using schema::GroupNode; - -namespace test { - -template <typename TestType> -class TestRowGroupStatistics : public PrimitiveTypedTest<TestType> { - public: - using T = typename TestType::c_type; - using TypedStats = TypedRowGroupStatistics<TestType>; - - std::vector<T> GetDeepCopy( - const std::vector<T>&); // allocates new memory for FLBA/ByteArray - - T* GetValuesPointer(std::vector<T>&); - void DeepFree(std::vector<T>&); - - void TestMinMaxEncode() { - this->GenerateData(1000); - - TypedStats statistics1(this->schema_.Column(0)); - statistics1.Update(this->values_ptr_, this->values_.size(), 0); - std::string encoded_min = statistics1.EncodeMin(); - std::string encoded_max = statistics1.EncodeMax(); - - TypedStats statistics2(this->schema_.Column(0), encoded_min, encoded_max, - this->values_.size(), 0, 0, true); - - TypedStats statistics3(this->schema_.Column(0)); - std::vector<uint8_t> valid_bits( - BitUtil::RoundUpNumBytes(static_cast<uint32_t>(this->values_.size())) + 1, 255); - statistics3.UpdateSpaced( - this->values_ptr_, valid_bits.data(), 0, this->values_.size(), 0); - std::string encoded_min_spaced = statistics3.EncodeMin(); - std::string encoded_max_spaced = statistics3.EncodeMax(); - - ASSERT_EQ(encoded_min, statistics2.EncodeMin()); - ASSERT_EQ(encoded_max, statistics2.EncodeMax()); - ASSERT_EQ(statistics1.min(), statistics2.min()); - ASSERT_EQ(statistics1.max(), statistics2.max()); - ASSERT_EQ(encoded_min_spaced, statistics2.EncodeMin()); - ASSERT_EQ(encoded_max_spaced, statistics2.EncodeMax()); - ASSERT_EQ(statistics3.min(), statistics2.min()); - ASSERT_EQ(statistics3.max(), statistics2.max()); - } - - void TestReset() { - this->GenerateData(1000); - - TypedStats statistics(this->schema_.Column(0)); - statistics.Update(this->values_ptr_, this->values_.size(), 0); - ASSERT_EQ(this->values_.size(), statistics.num_values()); - - statistics.Reset(); - ASSERT_EQ(0, statistics.null_count()); - ASSERT_EQ(0, statistics.num_values()); - ASSERT_EQ("", statistics.EncodeMin()); - ASSERT_EQ("", statistics.EncodeMax()); - } - - void TestMerge() { - int num_null[2]; - random_numbers(2, 42, 0, 100, num_null); - - TypedStats statistics1(this->schema_.Column(0)); - this->GenerateData(1000); - statistics1.Update( - this->values_ptr_, this->values_.size() - num_null[0], num_null[0]); - - TypedStats statistics2(this->schema_.Column(0)); - this->GenerateData(1000); - statistics2.Update( - this->values_ptr_, this->values_.size() - num_null[1], num_null[1]); - - TypedStats total(this->schema_.Column(0)); - total.Merge(statistics1); - total.Merge(statistics2); - - ASSERT_EQ(num_null[0] + num_null[1], total.null_count()); - ASSERT_EQ(this->values_.size() * 2 - num_null[0] - num_null[1], total.num_values()); - ASSERT_EQ(total.min(), std::min(statistics1.min(), statistics2.min())); - ASSERT_EQ(total.max(), std::max(statistics1.max(), statistics2.max())); - } - - void TestFullRoundtrip(int64_t num_values, int64_t null_count) { - this->GenerateData(num_values); - - // compute statistics for the whole batch - TypedStats expected_stats(this->schema_.Column(0)); - expected_stats.Update(this->values_ptr_, num_values - null_count, null_count); - - auto sink = std::make_shared<InMemoryOutputStream>(); - auto gnode = std::static_pointer_cast<GroupNode>(this->node_); - std::shared_ptr<WriterProperties> writer_properties = - WriterProperties::Builder().enable_statistics("column")->build(); - auto file_writer = ParquetFileWriter::Open(sink, gnode, writer_properties); - auto row_group_writer = file_writer->AppendRowGroup(num_values); - auto column_writer = - static_cast<TypedColumnWriter<TestType>*>(row_group_writer->NextColumn()); - - // simulate the case when data comes from multiple buffers, - // in which case special care is necessary for FLBA/ByteArray types - for (int i = 0; i < 2; i++) { - int64_t batch_num_values = i ? num_values - num_values / 2 : num_values / 2; - int64_t batch_null_count = i ? null_count : 0; - DCHECK(null_count <= num_values); // avoid too much headache - std::vector<int16_t> definition_levels(batch_null_count, 0); - definition_levels.insert( - definition_levels.end(), batch_num_values - batch_null_count, 1); - auto beg = this->values_.begin() + i * num_values / 2; - auto end = beg + batch_num_values; - std::vector<T> batch = GetDeepCopy(std::vector<T>(beg, end)); - T* batch_values_ptr = GetValuesPointer(batch); - column_writer->WriteBatch( - batch_num_values, definition_levels.data(), nullptr, batch_values_ptr); - DeepFree(batch); - } - column_writer->Close(); - row_group_writer->Close(); - file_writer->Close(); - - auto buffer = sink->GetBuffer(); - auto source = std::make_shared<::arrow::io::BufferReader>(buffer); - auto file_reader = ParquetFileReader::Open(source); - auto rg_reader = file_reader->RowGroup(0); - auto column_chunk = rg_reader->metadata()->ColumnChunk(0); - if (!column_chunk->is_stats_set()) return; - std::shared_ptr<RowGroupStatistics> stats = column_chunk->statistics(); - // check values after serialization + deserialization - ASSERT_EQ(null_count, stats->null_count()); - ASSERT_EQ(num_values - null_count, stats->num_values()); - ASSERT_EQ(expected_stats.EncodeMin(), stats->EncodeMin()); - ASSERT_EQ(expected_stats.EncodeMax(), stats->EncodeMax()); - } -}; - -template <typename TestType> -typename TestType::c_type* TestRowGroupStatistics<TestType>::GetValuesPointer( - std::vector<typename TestType::c_type>& values) { - return values.data(); -} - -template <> -bool* TestRowGroupStatistics<BooleanType>::GetValuesPointer(std::vector<bool>& values) { - static std::vector<uint8_t> bool_buffer; - bool_buffer.clear(); - bool_buffer.resize(values.size()); - std::copy(values.begin(), values.end(), bool_buffer.begin()); - return reinterpret_cast<bool*>(bool_buffer.data()); -} - -template <typename TestType> -typename std::vector<typename TestType::c_type> -TestRowGroupStatistics<TestType>::GetDeepCopy( - const std::vector<typename TestType::c_type>& values) { - return values; -} - -template <> -std::vector<FLBA> TestRowGroupStatistics<FLBAType>::GetDeepCopy( - const std::vector<FLBA>& values) { - std::vector<FLBA> copy; - MemoryPool* pool = ::arrow::default_memory_pool(); - for (const FLBA& flba : values) { - uint8_t* ptr; - PARQUET_THROW_NOT_OK(pool->Allocate(FLBA_LENGTH, &ptr)); - memcpy(ptr, flba.ptr, FLBA_LENGTH); - copy.emplace_back(ptr); - } - return copy; -} - -template <> -std::vector<ByteArray> TestRowGroupStatistics<ByteArrayType>::GetDeepCopy( - const std::vector<ByteArray>& values) { - std::vector<ByteArray> copy; - MemoryPool* pool = default_memory_pool(); - for (const ByteArray& ba : values) { - uint8_t* ptr; - PARQUET_THROW_NOT_OK(pool->Allocate(ba.len, &ptr)); - memcpy(ptr, ba.ptr, ba.len); - copy.emplace_back(ba.len, ptr); - } - return copy; -} - -template <typename TestType> -void TestRowGroupStatistics<TestType>::DeepFree( - std::vector<typename TestType::c_type>& values) {} - -template <> -void TestRowGroupStatistics<FLBAType>::DeepFree(std::vector<FLBA>& values) { - MemoryPool* pool = default_memory_pool(); - for (FLBA& flba : values) { - auto ptr = const_cast<uint8_t*>(flba.ptr); - memset(ptr, 0, FLBA_LENGTH); - pool->Free(ptr, FLBA_LENGTH); - } -} - -template <> -void TestRowGroupStatistics<ByteArrayType>::DeepFree(std::vector<ByteArray>& values) { - MemoryPool* pool = default_memory_pool(); - for (ByteArray& ba : values) { - auto ptr = const_cast<uint8_t*>(ba.ptr); - memset(ptr, 0, ba.len); - pool->Free(ptr, ba.len); - } -} - -template <> -void TestRowGroupStatistics<ByteArrayType>::TestMinMaxEncode() { - this->GenerateData(1000); - // Test that we encode min max strings correctly - TypedRowGroupStatistics<ByteArrayType> statistics1(this->schema_.Column(0)); - statistics1.Update(this->values_ptr_, this->values_.size(), 0); - std::string encoded_min = statistics1.EncodeMin(); - std::string encoded_max = statistics1.EncodeMax(); - - // encoded is same as unencoded - ASSERT_EQ(encoded_min, - std::string((const char*)statistics1.min().ptr, statistics1.min().len)); - ASSERT_EQ(encoded_max, - std::string((const char*)statistics1.max().ptr, statistics1.max().len)); - - TypedRowGroupStatistics<ByteArrayType> statistics2(this->schema_.Column(0), encoded_min, - encoded_max, this->values_.size(), 0, 0, true); - - ASSERT_EQ(encoded_min, statistics2.EncodeMin()); - ASSERT_EQ(encoded_max, statistics2.EncodeMax()); - ASSERT_EQ(statistics1.min(), statistics2.min()); - ASSERT_EQ(statistics1.max(), statistics2.max()); -} - -using TestTypes = ::testing::Types<Int32Type, Int64Type, Int96Type, FloatType, DoubleType, - ByteArrayType, FLBAType, BooleanType>; - -TYPED_TEST_CASE(TestRowGroupStatistics, TestTypes); - -TYPED_TEST(TestRowGroupStatistics, MinMaxEncode) { - this->SetUpSchema(Repetition::REQUIRED); - this->TestMinMaxEncode(); -} - -TYPED_TEST(TestRowGroupStatistics, Reset) { - this->SetUpSchema(Repetition::OPTIONAL); - this->TestReset(); -} - -TYPED_TEST(TestRowGroupStatistics, FullRoundtrip) { - this->SetUpSchema(Repetition::OPTIONAL); - this->TestFullRoundtrip(100, 31); - this->TestFullRoundtrip(1000, 415); - this->TestFullRoundtrip(10000, 926); -} - -template <typename TestType> -class TestNumericRowGroupStatistics : public TestRowGroupStatistics<TestType> {}; - -using NumericTypes = ::testing::Types<Int32Type, Int64Type, FloatType, DoubleType>; - -TYPED_TEST_CASE(TestNumericRowGroupStatistics, NumericTypes); - -TYPED_TEST(TestNumericRowGroupStatistics, Merge) { - this->SetUpSchema(Repetition::OPTIONAL); - this->TestMerge(); -} - -TEST(CorruptStatistics, Basics) { - ApplicationVersion version("parquet-mr version 1.8.0"); - SchemaDescriptor schema; - schema::NodePtr node; - std::vector<schema::NodePtr> fields; - // Test Physical Types - fields.push_back(schema::PrimitiveNode::Make( - "col1", Repetition::OPTIONAL, Type::INT32, LogicalType::NONE)); - fields.push_back(schema::PrimitiveNode::Make( - "col2", Repetition::OPTIONAL, Type::BYTE_ARRAY, LogicalType::NONE)); - // Test Logical Types - fields.push_back(schema::PrimitiveNode::Make( - "col3", Repetition::OPTIONAL, Type::INT32, LogicalType::DATE)); - fields.push_back(schema::PrimitiveNode::Make( - "col4", Repetition::OPTIONAL, Type::INT32, LogicalType::UINT_32)); - fields.push_back(schema::PrimitiveNode::Make("col5", Repetition::OPTIONAL, - Type::FIXED_LEN_BYTE_ARRAY, LogicalType::INTERVAL, 12)); - fields.push_back(schema::PrimitiveNode::Make( - "col6", Repetition::OPTIONAL, Type::BYTE_ARRAY, LogicalType::UTF8)); - node = schema::GroupNode::Make("schema", Repetition::REQUIRED, fields); - schema.Init(node); - - format::ColumnChunk col_chunk; - col_chunk.meta_data.__isset.statistics = true; - auto column_chunk1 = ColumnChunkMetaData::Make( - reinterpret_cast<const uint8_t*>(&col_chunk), schema.Column(0), &version); - ASSERT_TRUE(column_chunk1->is_stats_set()); - auto column_chunk2 = ColumnChunkMetaData::Make( - reinterpret_cast<const uint8_t*>(&col_chunk), schema.Column(1), &version); - ASSERT_FALSE(column_chunk2->is_stats_set()); - auto column_chunk3 = ColumnChunkMetaData::Make( - reinterpret_cast<const uint8_t*>(&col_chunk), schema.Column(2), &version); - ASSERT_TRUE(column_chunk3->is_stats_set()); - auto column_chunk4 = ColumnChunkMetaData::Make( - reinterpret_cast<const uint8_t*>(&col_chunk), schema.Column(3), &version); - ASSERT_FALSE(column_chunk4->is_stats_set()); - auto column_chunk5 = ColumnChunkMetaData::Make( - reinterpret_cast<const uint8_t*>(&col_chunk), schema.Column(4), &version); - ASSERT_FALSE(column_chunk5->is_stats_set()); - auto column_chunk6 = ColumnChunkMetaData::Make( - reinterpret_cast<const uint8_t*>(&col_chunk), schema.Column(5), &version); - ASSERT_FALSE(column_chunk6->is_stats_set()); -} - -} // namespace test -} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/84db929e/src/parquet/column/statistics.cc ---------------------------------------------------------------------- diff --git a/src/parquet/column/statistics.cc b/src/parquet/column/statistics.cc deleted file mode 100644 index 961a2af..0000000 --- a/src/parquet/column/statistics.cc +++ /dev/null @@ -1,244 +0,0 @@ -// 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 <algorithm> -#include <cstring> - -#include "parquet/column/statistics.h" -#include "parquet/encoding-internal.h" -#include "parquet/exception.h" -#include "parquet/util/comparison.h" -#include "parquet/util/memory.h" - -using arrow::default_memory_pool; -using arrow::MemoryPool; - -namespace parquet { - -template <typename DType> -TypedRowGroupStatistics<DType>::TypedRowGroupStatistics( - const ColumnDescriptor* schema, MemoryPool* pool) - : pool_(pool), - min_buffer_(AllocateBuffer(pool_, 0)), - max_buffer_(AllocateBuffer(pool_, 0)) { - SetDescr(schema); - Reset(); -} - -template <typename DType> -TypedRowGroupStatistics<DType>::TypedRowGroupStatistics(const typename DType::c_type& min, - const typename DType::c_type& max, int64_t num_values, int64_t null_count, - int64_t distinct_count) - : pool_(default_memory_pool()), - min_buffer_(AllocateBuffer(pool_, 0)), - max_buffer_(AllocateBuffer(pool_, 0)) { - IncrementNumValues(num_values); - IncrementNullCount(null_count); - IncrementDistinctCount(distinct_count); - - Copy(min, &min_, min_buffer_.get()); - Copy(max, &max_, max_buffer_.get()); - has_min_max_ = true; -} - -template <typename DType> -TypedRowGroupStatistics<DType>::TypedRowGroupStatistics(const ColumnDescriptor* schema, - const std::string& encoded_min, const std::string& encoded_max, int64_t num_values, - int64_t null_count, int64_t distinct_count, bool has_min_max, MemoryPool* pool) - : pool_(pool), - min_buffer_(AllocateBuffer(pool_, 0)), - max_buffer_(AllocateBuffer(pool_, 0)) { - IncrementNumValues(num_values); - IncrementNullCount(null_count); - IncrementDistinctCount(distinct_count); - - SetDescr(schema); - - if (!encoded_min.empty()) { PlainDecode(encoded_min, &min_); } - if (!encoded_max.empty()) { PlainDecode(encoded_max, &max_); } - has_min_max_ = has_min_max; -} - -template <typename DType> -bool TypedRowGroupStatistics<DType>::HasMinMax() const { - return has_min_max_; -} - -template <typename DType> -void TypedRowGroupStatistics<DType>::Reset() { - ResetCounts(); - has_min_max_ = false; -} - -template <typename DType> -void TypedRowGroupStatistics<DType>::Update( - const T* values, int64_t num_not_null, int64_t num_null) { - DCHECK(num_not_null >= 0); - DCHECK(num_null >= 0); - - IncrementNullCount(num_null); - IncrementNumValues(num_not_null); - // TODO: support distinct count? - if (num_not_null == 0) return; - - Compare<T> compare(descr_); - auto batch_minmax = std::minmax_element(values, values + num_not_null, compare); - if (!has_min_max_) { - has_min_max_ = true; - Copy(*batch_minmax.first, &min_, min_buffer_.get()); - Copy(*batch_minmax.second, &max_, max_buffer_.get()); - } else { - Copy(std::min(min_, *batch_minmax.first, compare), &min_, min_buffer_.get()); - Copy(std::max(max_, *batch_minmax.second, compare), &max_, max_buffer_.get()); - } -} - -template <typename DType> -void TypedRowGroupStatistics<DType>::UpdateSpaced(const T* values, - const uint8_t* valid_bits, int64_t valid_bits_offset, int64_t num_not_null, - int64_t num_null) { - DCHECK(num_not_null >= 0); - DCHECK(num_null >= 0); - - IncrementNullCount(num_null); - IncrementNumValues(num_not_null); - // TODO: support distinct count? - if (num_not_null == 0) return; - - Compare<T> compare(descr_); - INIT_BITSET(valid_bits, static_cast<int>(valid_bits_offset)); - // Find first valid entry and use that for min/max - // As (num_not_null != 0) there must be one - int64_t length = num_null + num_not_null; - int64_t i = 0; - for (; i < length; i++) { - if (bitset_valid_bits & (1 << bit_offset_valid_bits)) { break; } - READ_NEXT_BITSET(valid_bits); - } - T min = values[i]; - T max = values[i]; - for (; i < length; i++) { - if (bitset_valid_bits & (1 << bit_offset_valid_bits)) { - if (compare(values[i], min)) { - min = values[i]; - } else if (compare(max, values[i])) { - max = values[i]; - } - } - READ_NEXT_BITSET(valid_bits); - } - if (!has_min_max_) { - has_min_max_ = true; - Copy(min, &min_, min_buffer_.get()); - Copy(max, &max_, max_buffer_.get()); - } else { - Copy(std::min(min_, min, compare), &min_, min_buffer_.get()); - Copy(std::max(max_, max, compare), &max_, max_buffer_.get()); - } -} - -template <typename DType> -const typename DType::c_type& TypedRowGroupStatistics<DType>::min() const { - return min_; -} - -template <typename DType> -const typename DType::c_type& TypedRowGroupStatistics<DType>::max() const { - return max_; -} - -template <typename DType> -void TypedRowGroupStatistics<DType>::Merge(const TypedRowGroupStatistics<DType>& other) { - this->MergeCounts(other); - - if (!other.HasMinMax()) return; - - if (!has_min_max_) { - Copy(other.min_, &this->min_, min_buffer_.get()); - Copy(other.max_, &this->max_, max_buffer_.get()); - has_min_max_ = true; - return; - } - - Compare<T> compare(descr_); - Copy(std::min(this->min_, other.min_, compare), &this->min_, min_buffer_.get()); - Copy(std::max(this->max_, other.max_, compare), &this->max_, max_buffer_.get()); -} - -template <typename DType> -std::string TypedRowGroupStatistics<DType>::EncodeMin() { - std::string s; - if (HasMinMax()) this->PlainEncode(min_, &s); - return s; -} - -template <typename DType> -std::string TypedRowGroupStatistics<DType>::EncodeMax() { - std::string s; - if (HasMinMax()) this->PlainEncode(max_, &s); - return s; -} - -template <typename DType> -EncodedStatistics TypedRowGroupStatistics<DType>::Encode() { - EncodedStatistics s; - if (HasMinMax()) { - s.set_min(this->EncodeMin()); - s.set_max(this->EncodeMax()); - } - s.set_null_count(this->null_count()); - return s; -} - -template <typename DType> -void TypedRowGroupStatistics<DType>::PlainEncode(const T& src, std::string* dst) { - PlainEncoder<DType> encoder(descr(), pool_); - encoder.Put(&src, 1); - auto buffer = encoder.FlushValues(); - auto ptr = reinterpret_cast<const char*>(buffer->data()); - dst->assign(ptr, buffer->size()); -} - -template <typename DType> -void TypedRowGroupStatistics<DType>::PlainDecode(const std::string& src, T* dst) { - PlainDecoder<DType> decoder(descr()); - decoder.SetData( - 1, reinterpret_cast<const uint8_t*>(src.c_str()), static_cast<int>(src.size())); - decoder.Decode(dst, 1); -} - -template <> -void TypedRowGroupStatistics<ByteArrayType>::PlainEncode(const T& src, std::string* dst) { - dst->assign(reinterpret_cast<const char*>(src.ptr), src.len); -} - -template <> -void TypedRowGroupStatistics<ByteArrayType>::PlainDecode(const std::string& src, T* dst) { - dst->len = static_cast<uint32_t>(src.size()); - dst->ptr = reinterpret_cast<const uint8_t*>(src.c_str()); -} - -template class PARQUET_TEMPLATE_EXPORT TypedRowGroupStatistics<BooleanType>; -template class PARQUET_TEMPLATE_EXPORT TypedRowGroupStatistics<Int32Type>; -template class PARQUET_TEMPLATE_EXPORT TypedRowGroupStatistics<Int64Type>; -template class PARQUET_TEMPLATE_EXPORT TypedRowGroupStatistics<Int96Type>; -template class PARQUET_TEMPLATE_EXPORT TypedRowGroupStatistics<FloatType>; -template class PARQUET_TEMPLATE_EXPORT TypedRowGroupStatistics<DoubleType>; -template class PARQUET_TEMPLATE_EXPORT TypedRowGroupStatistics<ByteArrayType>; -template class PARQUET_TEMPLATE_EXPORT TypedRowGroupStatistics<FLBAType>; - -} // namespace parquet http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/84db929e/src/parquet/column/statistics.h ---------------------------------------------------------------------- diff --git a/src/parquet/column/statistics.h b/src/parquet/column/statistics.h deleted file mode 100644 index c6a2487..0000000 --- a/src/parquet/column/statistics.h +++ /dev/null @@ -1,234 +0,0 @@ -// 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. - -#ifndef PARQUET_COLUMN_STATISTICS_H -#define PARQUET_COLUMN_STATISTICS_H - -#include <cstdint> -#include <memory> -#include <string> - -#include "parquet/schema.h" -#include "parquet/types.h" -#include "parquet/util/memory.h" -#include "parquet/util/visibility.h" - -namespace parquet { - -class PARQUET_EXPORT EncodedStatistics { - std::shared_ptr<std::string> max_, min_; - - public: - EncodedStatistics() - : max_(std::make_shared<std::string>()), min_(std::make_shared<std::string>()) {} - - const std::string& max() const { return *max_; } - const std::string& min() const { return *min_; } - - int64_t null_count = 0; - int64_t distinct_count = 0; - - bool has_min = false; - bool has_max = false; - bool has_null_count = false; - bool has_distinct_count = false; - - inline bool is_set() const { - return has_min || has_max || has_null_count || has_distinct_count; - } - - inline EncodedStatistics& set_max(const std::string& value) { - *max_ = value; - has_max = true; - return *this; - } - - inline EncodedStatistics& set_min(const std::string& value) { - *min_ = value; - has_min = true; - return *this; - } - - inline EncodedStatistics& set_null_count(int64_t value) { - null_count = value; - has_null_count = true; - return *this; - } - - inline EncodedStatistics& set_distinct_count(int64_t value) { - distinct_count = value; - has_distinct_count = true; - return *this; - } -}; - -template <typename DType> -class PARQUET_EXPORT TypedRowGroupStatistics; - -class PARQUET_EXPORT RowGroupStatistics - : public std::enable_shared_from_this<RowGroupStatistics> { - public: - int64_t null_count() const { return statistics_.null_count; } - int64_t distinct_count() const { return statistics_.distinct_count; } - int64_t num_values() const { return num_values_; } - - virtual bool HasMinMax() const = 0; - virtual void Reset() = 0; - - // Plain-encoded minimum value - virtual std::string EncodeMin() = 0; - - // Plain-encoded maximum value - virtual std::string EncodeMax() = 0; - - virtual EncodedStatistics Encode() = 0; - - virtual ~RowGroupStatistics() {} - - Type::type physical_type() const { return descr_->physical_type(); } - - protected: - const ColumnDescriptor* descr() const { return descr_; } - void SetDescr(const ColumnDescriptor* schema) { descr_ = schema; } - - void IncrementNullCount(int64_t n) { statistics_.null_count += n; } - - void IncrementNumValues(int64_t n) { num_values_ += n; } - - void IncrementDistinctCount(int64_t n) { statistics_.distinct_count += n; } - - void MergeCounts(const RowGroupStatistics& other) { - this->statistics_.null_count += other.statistics_.null_count; - this->statistics_.distinct_count += other.statistics_.distinct_count; - this->num_values_ += other.num_values_; - } - - void ResetCounts() { - this->statistics_.null_count = 0; - this->statistics_.distinct_count = 0; - this->num_values_ = 0; - } - - const ColumnDescriptor* descr_ = nullptr; - int64_t num_values_ = 0; - EncodedStatistics statistics_; -}; - -template <typename DType> -class TypedRowGroupStatistics : public RowGroupStatistics { - public: - using T = typename DType::c_type; - - TypedRowGroupStatistics(const ColumnDescriptor* schema, - ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()); - - TypedRowGroupStatistics(const T& min, const T& max, int64_t num_values, - int64_t null_count, int64_t distinct_count); - - TypedRowGroupStatistics(const ColumnDescriptor* schema, const std::string& encoded_min, - const std::string& encoded_max, int64_t num_values, int64_t null_count, - int64_t distinct_count, bool has_min_max, - ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()); - - bool HasMinMax() const override; - void Reset() override; - void Merge(const TypedRowGroupStatistics<DType>& other); - - void Update(const T* values, int64_t num_not_null, int64_t num_null); - void UpdateSpaced(const T* values, const uint8_t* valid_bits, int64_t valid_bits_spaced, - int64_t num_not_null, int64_t num_null); - - const T& min() const; - const T& max() const; - - std::string EncodeMin() override; - std::string EncodeMax() override; - EncodedStatistics Encode() override; - - private: - bool has_min_max_ = false; - T min_; - T max_; - ::arrow::MemoryPool* pool_; - - void PlainEncode(const T& src, std::string* dst); - void PlainDecode(const std::string& src, T* dst); - void Copy(const T& src, T* dst, PoolBuffer* buffer); - - std::shared_ptr<PoolBuffer> min_buffer_, max_buffer_; -}; - -template <typename DType> -inline void TypedRowGroupStatistics<DType>::Copy(const T& src, T* dst, PoolBuffer*) { - *dst = src; -} - -template <> -inline void TypedRowGroupStatistics<FLBAType>::Copy( - const FLBA& src, FLBA* dst, PoolBuffer* buffer) { - if (dst->ptr == src.ptr) return; - uint32_t len = descr_->type_length(); - PARQUET_THROW_NOT_OK(buffer->Resize(len, false)); - std::memcpy(buffer->mutable_data(), src.ptr, len); - *dst = FLBA(buffer->data()); -} - -template <> -inline void TypedRowGroupStatistics<ByteArrayType>::Copy( - const ByteArray& src, ByteArray* dst, PoolBuffer* buffer) { - if (dst->ptr == src.ptr) return; - PARQUET_THROW_NOT_OK(buffer->Resize(src.len, false)); - std::memcpy(buffer->mutable_data(), src.ptr, src.len); - *dst = ByteArray(src.len, buffer->data()); -} - -template <> -void TypedRowGroupStatistics<ByteArrayType>::PlainEncode(const T& src, std::string* dst); - -template <> -void TypedRowGroupStatistics<ByteArrayType>::PlainDecode(const std::string& src, T* dst); - -typedef TypedRowGroupStatistics<BooleanType> BoolStatistics; -typedef TypedRowGroupStatistics<Int32Type> Int32Statistics; -typedef TypedRowGroupStatistics<Int64Type> Int64Statistics; -typedef TypedRowGroupStatistics<Int96Type> Int96Statistics; -typedef TypedRowGroupStatistics<FloatType> FloatStatistics; -typedef TypedRowGroupStatistics<DoubleType> DoubleStatistics; -typedef TypedRowGroupStatistics<ByteArrayType> ByteArrayStatistics; -typedef TypedRowGroupStatistics<FLBAType> FLBAStatistics; - -#if defined(__GNUC__) && !defined(__clang__) -#pragma GCC diagnostic push -#pragma GCC diagnostic ignored "-Wattributes" -#endif - -PARQUET_EXTERN_TEMPLATE TypedRowGroupStatistics<BooleanType>; -PARQUET_EXTERN_TEMPLATE TypedRowGroupStatistics<Int32Type>; -PARQUET_EXTERN_TEMPLATE TypedRowGroupStatistics<Int64Type>; -PARQUET_EXTERN_TEMPLATE TypedRowGroupStatistics<Int96Type>; -PARQUET_EXTERN_TEMPLATE TypedRowGroupStatistics<FloatType>; -PARQUET_EXTERN_TEMPLATE TypedRowGroupStatistics<DoubleType>; -PARQUET_EXTERN_TEMPLATE TypedRowGroupStatistics<ByteArrayType>; -PARQUET_EXTERN_TEMPLATE TypedRowGroupStatistics<FLBAType>; - -#if defined(__GNUC__) && !defined(__clang__) -#pragma GCC diagnostic pop -#endif - -} // namespace parquet - -#endif // PARQUET_COLUMN_STATISTICS_H http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/84db929e/src/parquet/column/test-specialization.h ---------------------------------------------------------------------- diff --git a/src/parquet/column/test-specialization.h b/src/parquet/column/test-specialization.h deleted file mode 100644 index 07767c0..0000000 --- a/src/parquet/column/test-specialization.h +++ /dev/null @@ -1,172 +0,0 @@ -// 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. - -// This module defines an abstract interface for iterating through pages in a -// Parquet column chunk within a row group. It could be extended in the future -// to iterate through all data pages in all chunks in a file. - -#ifndef PARQUET_COLUMN_TEST_SPECIALIZATION_H -#define PARQUET_COLUMN_TEST_SPECIALIZATION_H - -#include <algorithm> -#include <limits> -#include <sstream> -#include <string> -#include <vector> - -#include "parquet/column/test-util.h" - -namespace parquet { - -namespace test { - -template <> -void InitValues<bool>(int num_values, vector<bool>& values, vector<uint8_t>& buffer) { - values = flip_coins(num_values, 0); -} - -template <> -void InitValues<ByteArray>( - int num_values, vector<ByteArray>& values, vector<uint8_t>& buffer) { - int max_byte_array_len = 12; - int num_bytes = max_byte_array_len + sizeof(uint32_t); - size_t nbytes = num_values * num_bytes; - buffer.resize(nbytes); - random_byte_array(num_values, 0, buffer.data(), values.data(), max_byte_array_len); -} - -template <> -void InitValues<FLBA>(int num_values, vector<FLBA>& values, vector<uint8_t>& buffer) { - size_t nbytes = num_values * FLBA_LENGTH; - buffer.resize(nbytes); - random_fixed_byte_array(num_values, 0, buffer.data(), FLBA_LENGTH, values.data()); -} - -template <> -void InitValues<Int96>(int num_values, vector<Int96>& values, vector<uint8_t>& buffer) { - random_Int96_numbers(num_values, 0, std::numeric_limits<int32_t>::min(), - std::numeric_limits<int32_t>::max(), values.data()); -} - -inline std::string TestColumnName(int i) { - std::stringstream col_name; - col_name << "column_" << i; - return col_name.str(); -} - -// This class lives here because of its dependency on the InitValues specializations. -template <typename TestType> -class PrimitiveTypedTest : public ::testing::Test { - public: - typedef typename TestType::c_type T; - - void SetUpSchema(Repetition::type repetition, int num_columns = 1) { - std::vector<schema::NodePtr> fields; - - for (int i = 0; i < num_columns; ++i) { - std::string name = TestColumnName(i); - fields.push_back(schema::PrimitiveNode::Make( - name, repetition, TestType::type_num, LogicalType::NONE, FLBA_LENGTH)); - } - node_ = schema::GroupNode::Make("schema", Repetition::REQUIRED, fields); - schema_.Init(node_); - } - - void GenerateData(int64_t num_values); - void SetupValuesOut(int64_t num_values); - void SyncValuesOut(); - - protected: - schema::NodePtr node_; - SchemaDescriptor schema_; - - // Input buffers - std::vector<T> values_; - - std::vector<int16_t> def_levels_; - - std::vector<uint8_t> buffer_; - // Pointer to the values, needed as we cannot use vector<bool>::data() - T* values_ptr_; - std::vector<uint8_t> bool_buffer_; - - // Output buffers - std::vector<T> values_out_; - std::vector<uint8_t> bool_buffer_out_; - T* values_out_ptr_; -}; - -template <typename TestType> -void PrimitiveTypedTest<TestType>::SyncValuesOut() {} - -template <> -void PrimitiveTypedTest<BooleanType>::SyncValuesOut() { - std::vector<uint8_t>::const_iterator source_iterator = bool_buffer_out_.begin(); - std::vector<T>::iterator destination_iterator = values_out_.begin(); - while (source_iterator != bool_buffer_out_.end()) { - *destination_iterator++ = *source_iterator++ != 0; - } -} - -template <typename TestType> -void PrimitiveTypedTest<TestType>::SetupValuesOut(int64_t num_values) { - values_out_.clear(); - values_out_.resize(num_values); - values_out_ptr_ = values_out_.data(); -} - -template <> -void PrimitiveTypedTest<BooleanType>::SetupValuesOut(int64_t num_values) { - values_out_.clear(); - values_out_.resize(num_values); - - bool_buffer_out_.clear(); - bool_buffer_out_.resize(num_values); - // Write once to all values so we can copy it without getting Valgrind errors - // about uninitialised values. - std::fill(bool_buffer_out_.begin(), bool_buffer_out_.end(), true); - values_out_ptr_ = reinterpret_cast<bool*>(bool_buffer_out_.data()); -} - -template <typename TestType> -void PrimitiveTypedTest<TestType>::GenerateData(int64_t num_values) { - def_levels_.resize(num_values); - values_.resize(num_values); - - InitValues<T>(static_cast<int>(num_values), values_, buffer_); - values_ptr_ = values_.data(); - - std::fill(def_levels_.begin(), def_levels_.end(), 1); -} - -template <> -void PrimitiveTypedTest<BooleanType>::GenerateData(int64_t num_values) { - def_levels_.resize(num_values); - values_.resize(num_values); - - InitValues<T>(static_cast<int>(num_values), values_, buffer_); - bool_buffer_.resize(num_values); - std::copy(values_.begin(), values_.end(), bool_buffer_.begin()); - values_ptr_ = reinterpret_cast<bool*>(bool_buffer_.data()); - - std::fill(def_levels_.begin(), def_levels_.end(), 1); -} -} // namespace test - -} // namespace parquet - -#endif // PARQUET_COLUMN_TEST_SPECIALIZATION_H http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/84db929e/src/parquet/column/test-util.h ---------------------------------------------------------------------- diff --git a/src/parquet/column/test-util.h b/src/parquet/column/test-util.h deleted file mode 100644 index c133734..0000000 --- a/src/parquet/column/test-util.h +++ /dev/null @@ -1,429 +0,0 @@ -// 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. - -// This module defines an abstract interface for iterating through pages in a -// Parquet column chunk within a row group. It could be extended in the future -// to iterate through all data pages in all chunks in a file. - -#ifndef PARQUET_COLUMN_TEST_UTIL_H -#define PARQUET_COLUMN_TEST_UTIL_H - -#include <algorithm> -#include <limits> -#include <memory> -#include <string> -#include <vector> - -#include <gtest/gtest.h> - -#include "parquet/column/levels.h" -#include "parquet/column/page.h" -#include "parquet/encoding-internal.h" -#include "parquet/util/memory.h" -#include "parquet/util/test-common.h" - -using std::vector; -using std::shared_ptr; - -namespace parquet { - -static int FLBA_LENGTH = 12; - -bool operator==(const FixedLenByteArray& a, const FixedLenByteArray& b) { - return 0 == memcmp(a.ptr, b.ptr, FLBA_LENGTH); -} - -namespace test { - -template <typename T> -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) { - int repeat_factor = num_values / num_dicts; - InitValues<T>(num_dicts, values, buffer); - // add some repeated values - for (int j = 1; j < repeat_factor; ++j) { - for (int i = 0; i < num_dicts; ++i) { - std::memcpy(&values[num_dicts * j + i], &values[i], sizeof(T)); - } - } - // computed only dict_per_page * repeat_factor - 1 values < num_values - // compute remaining - for (int i = num_dicts * repeat_factor; i < num_values; ++i) { - std::memcpy(&values[i], &values[i - num_dicts * repeat_factor], sizeof(T)); - } -} - -class MockPageReader : public PageReader { - public: - explicit MockPageReader(const vector<shared_ptr<Page>>& pages) - : pages_(pages), page_index_(0) {} - - // Implement the PageReader interface - virtual shared_ptr<Page> NextPage() { - if (page_index_ == static_cast<int>(pages_.size())) { - // EOS to consumer - return shared_ptr<Page>(nullptr); - } - return pages_[page_index_++]; - } - - private: - vector<shared_ptr<Page>> pages_; - int page_index_; -}; - -// TODO(wesm): this is only used for testing for now. Refactor to form part of -// primary file write path -template <typename Type> -class DataPageBuilder { - public: - 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) {} - - void AppendDefLevels(const vector<int16_t>& levels, int16_t max_level, - Encoding::type encoding = Encoding::RLE) { - AppendLevels(levels, max_level, encoding); - - num_values_ = std::max(static_cast<int32_t>(levels.size()), num_values_); - definition_level_encoding_ = encoding; - have_def_levels_ = true; - } - - void AppendRepLevels(const vector<int16_t>& levels, int16_t max_level, - Encoding::type encoding = Encoding::RLE) { - AppendLevels(levels, max_level, encoding); - - num_values_ = std::max(static_cast<int32_t>(levels.size()), num_values_); - repetition_level_encoding_ = encoding; - have_rep_levels_ = true; - } - - void AppendValues(const ColumnDescriptor* d, const vector<T>& values, - Encoding::type encoding = Encoding::PLAIN) { - PlainEncoder<Type> encoder(d); - encoder.Put(&values[0], static_cast<int>(values.size())); - std::shared_ptr<Buffer> values_sink = encoder.FlushValues(); - sink_->Write(values_sink->data(), values_sink->size()); - - num_values_ = std::max(static_cast<int32_t>(values.size()), num_values_); - encoding_ = encoding; - have_values_ = true; - } - - int32_t num_values() const { return num_values_; } - - Encoding::type encoding() const { return encoding_; } - - Encoding::type rep_level_encoding() const { return repetition_level_encoding_; } - - Encoding::type def_level_encoding() const { return definition_level_encoding_; } - - private: - InMemoryOutputStream* sink_; - - int32_t num_values_; - Encoding::type encoding_; - Encoding::type definition_level_encoding_; - Encoding::type repetition_level_encoding_; - - bool have_def_levels_; - bool have_rep_levels_; - 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) { - if (encoding != Encoding::RLE) { - ParquetException::NYI("only rle encoding currently implemented"); - } - - // TODO: compute a more precise maximum size for the encoded levels - vector<uint8_t> encode_buffer(levels.size() * 2); - - // We encode into separate memory from the output stream because the - // RLE-encoded bytes have to be preceded in the stream by their absolute - // size. - LevelEncoder encoder; - encoder.Init(encoding, max_level, static_cast<int>(levels.size()), - encode_buffer.data(), static_cast<int>(encode_buffer.size())); - - encoder.Encode(static_cast<int>(levels.size()), levels.data()); - - int32_t rle_bytes = encoder.len(); - sink_->Write(reinterpret_cast<const uint8_t*>(&rle_bytes), sizeof(int32_t)); - sink_->Write(encode_buffer.data(), rle_bytes); - } -}; - -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"); - } - PlainEncoder<BooleanType> encoder(d); - encoder.Put(values, static_cast<int>(values.size())); - std::shared_ptr<Buffer> buffer = encoder.FlushValues(); - sink_->Write(buffer->data(), buffer->size()); - - num_values_ = std::max(static_cast<int32_t>(values.size()), num_values_); - encoding_ = encoding; - have_values_ = true; -} - -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) { - 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 (encoding == Encoding::PLAIN) { - page_builder.AppendValues(d, values, encoding); - num_values = page_builder.num_values(); - } 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()); -} - -template <typename TYPE> -class DictionaryPageBuilder { - public: - typedef typename TYPE::c_type TC; - 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) { - encoder_.reset(new DictEncoder<TYPE>(d, &pool_)); - } - - ~DictionaryPageBuilder() { pool_.FreeAll(); } - - shared_ptr<Buffer> AppendValues(const vector<TC>& values) { - int num_values = static_cast<int>(values.size()); - // Dictionary encoding - encoder_->Put(values.data(), num_values); - num_dict_values_ = encoder_->num_entries(); - have_values_ = true; - return encoder_->FlushValues(); - } - - shared_ptr<Buffer> WriteDict() { - std::shared_ptr<PoolBuffer> dict_buffer = - AllocateBuffer(::arrow::default_memory_pool(), encoder_->dict_encoded_size()); - encoder_->WriteDict(dict_buffer->mutable_data()); - return dict_buffer; - } - - int32_t num_values() const { return num_dict_values_; } - - private: - ChunkedAllocator pool_; - shared_ptr<DictEncoder<TYPE>> encoder_; - int32_t num_dict_values_; - bool have_values_; -}; - -template <> -DictionaryPageBuilder<BooleanType>::DictionaryPageBuilder(const ColumnDescriptor* d) { - ParquetException::NYI("only plain encoding currently implemented for boolean"); -} - -template <> -shared_ptr<Buffer> DictionaryPageBuilder<BooleanType>::WriteDict() { - ParquetException::NYI("only plain encoding currently implemented for boolean"); - return nullptr; -} - -template <> -shared_ptr<Buffer> DictionaryPageBuilder<BooleanType>::AppendValues( - const vector<TC>& values) { - ParquetException::NYI("only plain encoding currently implemented for boolean"); - return nullptr; -} - -template <typename Type> -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) { - InMemoryOutputStream page_stream; - test::DictionaryPageBuilder<Type> page_builder(d); - int num_pages = static_cast<int>(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]))); - value_start += values_per_page[i]; - } - - auto buffer = page_builder.WriteDict(); - - 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, - int num_levels_per_page, const vector<int>& values_per_page, - vector<shared_ptr<Page>>& pages, Encoding::type encoding = Encoding::RLE_DICTIONARY) { - int num_pages = static_cast<int>(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); - pages.push_back(dict_page); - int def_level_start = 0; - int def_level_end = 0; - int rep_level_start = 0; - int rep_level_end = 0; - for (int i = 0; i < num_pages; i++) { - if (max_def_level > 0) { - def_level_start = i * num_levels_per_page; - def_level_end = (i + 1) * num_levels_per_page; - } - if (max_rep_level > 0) { - rep_level_start = i * num_levels_per_page; - rep_level_end = (i + 1) * num_levels_per_page; - } - shared_ptr<DataPage> data_page = MakeDataPage<Int32Type>(d, {}, values_per_page[i], - encoding, rle_indices[i]->data(), static_cast<int>(rle_indices[i]->size()), - 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(data_page); - } -} - -// 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, - int num_levels_per_page, const vector<int>& values_per_page, - vector<shared_ptr<Page>>& pages, Encoding::type encoding = Encoding::PLAIN) { - int num_pages = static_cast<int>(values_per_page.size()); - int def_level_start = 0; - int def_level_end = 0; - int rep_level_start = 0; - int rep_level_end = 0; - int value_start = 0; - for (int i = 0; i < num_pages; i++) { - if (max_def_level > 0) { - def_level_start = i * num_levels_per_page; - def_level_end = (i + 1) * num_levels_per_page; - } - if (max_rep_level > 0) { - rep_level_start = i * num_levels_per_page; - rep_level_end = (i + 1) * num_levels_per_page; - } - 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); - pages.push_back(page); - value_start += values_per_page[i]; - } -} - -// Generates pages from randomly generated data -template <typename Type> -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) { - int num_levels = levels_per_page * num_pages; - int num_values = 0; - uint32_t seed = 0; - int16_t zero = 0; - int16_t max_def_level = d->max_definition_level(); - int16_t max_rep_level = d->max_repetition_level(); - vector<int> values_per_page(num_pages, levels_per_page); - // Create definition levels - if (max_def_level > 0) { - def_levels.resize(num_levels); - random_numbers(num_levels, seed, zero, max_def_level, def_levels.data()); - for (int p = 0; p < num_pages; p++) { - int num_values_per_page = 0; - for (int i = 0; i < levels_per_page; i++) { - if (def_levels[i + p * levels_per_page] == max_def_level) { - num_values_per_page++; - num_values++; - } - } - values_per_page[p] = num_values_per_page; - } - } else { - num_values = num_levels; - } - // Create repitition levels - if (max_rep_level > 0) { - rep_levels.resize(num_levels); - random_numbers(num_levels, seed, zero, max_rep_level, rep_levels.data()); - } - // Create values - 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) { - // 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); - } - - return num_values; -} - -} // namespace test - -} // namespace parquet - -#endif // PARQUET_COLUMN_TEST_UTIL_H http://git-wip-us.apache.org/repos/asf/parquet-cpp/blob/84db929e/src/parquet/column/writer.cc ---------------------------------------------------------------------- diff --git a/src/parquet/column/writer.cc b/src/parquet/column/writer.cc deleted file mode 100644 index 59f9999..0000000 --- a/src/parquet/column/writer.cc +++ /dev/null @@ -1,528 +0,0 @@ -// 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 "parquet/column/writer.h" - -#include "parquet/column/properties.h" -#include "parquet/column/statistics.h" -#include "parquet/encoding-internal.h" -#include "parquet/util/logging.h" -#include "parquet/util/memory.h" - -namespace parquet { - -// ---------------------------------------------------------------------- -// ColumnWriter - -std::shared_ptr<WriterProperties> default_writer_properties() { - static std::shared_ptr<WriterProperties> default_writer_properties = - WriterProperties::Builder().build(); - return default_writer_properties; -} - -ColumnWriter::ColumnWriter(ColumnChunkMetaDataBuilder* metadata, - std::unique_ptr<PageWriter> pager, int64_t expected_rows, bool has_dictionary, - Encoding::type encoding, const WriterProperties* properties) - : metadata_(metadata), - descr_(metadata->descr()), - pager_(std::move(pager)), - expected_rows_(expected_rows), - has_dictionary_(has_dictionary), - encoding_(encoding), - properties_(properties), - allocator_(properties->memory_pool()), - pool_(properties->memory_pool()), - num_buffered_values_(0), - num_buffered_encoded_values_(0), - num_rows_(0), - total_bytes_written_(0), - closed_(false), - fallback_(false) { - definition_levels_sink_.reset(new InMemoryOutputStream(allocator_)); - repetition_levels_sink_.reset(new InMemoryOutputStream(allocator_)); - definition_levels_rle_ = - std::static_pointer_cast<ResizableBuffer>(AllocateBuffer(allocator_, 0)); - repetition_levels_rle_ = - std::static_pointer_cast<ResizableBuffer>(AllocateBuffer(allocator_, 0)); - uncompressed_data_ = - std::static_pointer_cast<ResizableBuffer>(AllocateBuffer(allocator_, 0)); - if (pager_->has_compressor()) { - compressed_data_ = - std::static_pointer_cast<ResizableBuffer>(AllocateBuffer(allocator_, 0)); - } -} - -void ColumnWriter::InitSinks() { - definition_levels_sink_->Clear(); - repetition_levels_sink_->Clear(); -} - -void ColumnWriter::WriteDefinitionLevels(int64_t num_levels, const int16_t* levels) { - DCHECK(!closed_); - definition_levels_sink_->Write( - reinterpret_cast<const uint8_t*>(levels), sizeof(int16_t) * num_levels); -} - -void ColumnWriter::WriteRepetitionLevels(int64_t num_levels, const int16_t* levels) { - DCHECK(!closed_); - repetition_levels_sink_->Write( - reinterpret_cast<const uint8_t*>(levels), sizeof(int16_t) * num_levels); -} - -// return the size of the encoded buffer -int64_t ColumnWriter::RleEncodeLevels( - const Buffer& src_buffer, ResizableBuffer* dest_buffer, int16_t max_level) { - // TODO: This only works with due to some RLE specifics - int64_t rle_size = LevelEncoder::MaxBufferSize(Encoding::RLE, max_level, - static_cast<int>(num_buffered_values_)) + - sizeof(int32_t); - - // Use Arrow::Buffer::shrink_to_fit = false - // underlying buffer only keeps growing. Resize to a smaller size does not reallocate. - PARQUET_THROW_NOT_OK(dest_buffer->Resize(rle_size, false)); - - level_encoder_.Init(Encoding::RLE, max_level, static_cast<int>(num_buffered_values_), - dest_buffer->mutable_data() + sizeof(int32_t), - static_cast<int>(dest_buffer->size()) - sizeof(int32_t)); - int encoded = level_encoder_.Encode(static_cast<int>(num_buffered_values_), - reinterpret_cast<const int16_t*>(src_buffer.data())); - DCHECK_EQ(encoded, num_buffered_values_); - reinterpret_cast<int32_t*>(dest_buffer->mutable_data())[0] = level_encoder_.len(); - int64_t encoded_size = level_encoder_.len() + sizeof(int32_t); - return encoded_size; -} - -void ColumnWriter::AddDataPage() { - int64_t definition_levels_rle_size = 0; - int64_t repetition_levels_rle_size = 0; - - std::shared_ptr<Buffer> values = GetValuesBuffer(); - - if (descr_->max_definition_level() > 0) { - definition_levels_rle_size = RleEncodeLevels(definition_levels_sink_->GetBufferRef(), - definition_levels_rle_.get(), descr_->max_definition_level()); - } - - if (descr_->max_repetition_level() > 0) { - repetition_levels_rle_size = RleEncodeLevels(repetition_levels_sink_->GetBufferRef(), - repetition_levels_rle_.get(), descr_->max_repetition_level()); - } - - int64_t uncompressed_size = - definition_levels_rle_size + repetition_levels_rle_size + values->size(); - - // Use Arrow::Buffer::shrink_to_fit = false - // underlying buffer only keeps growing. Resize to a smaller size does not reallocate. - PARQUET_THROW_NOT_OK(uncompressed_data_->Resize(uncompressed_size, false)); - - // Concatenate data into a single buffer - uint8_t* uncompressed_ptr = uncompressed_data_->mutable_data(); - memcpy(uncompressed_ptr, repetition_levels_rle_->data(), repetition_levels_rle_size); - uncompressed_ptr += repetition_levels_rle_size; - memcpy(uncompressed_ptr, definition_levels_rle_->data(), definition_levels_rle_size); - uncompressed_ptr += definition_levels_rle_size; - memcpy(uncompressed_ptr, values->data(), values->size()); - - EncodedStatistics page_stats = GetPageStatistics(); - ResetPageStatistics(); - - std::shared_ptr<Buffer> compressed_data; - if (pager_->has_compressor()) { - pager_->Compress(*(uncompressed_data_.get()), compressed_data_.get()); - compressed_data = compressed_data_; - } else { - compressed_data = uncompressed_data_; - } - - // Write the page to OutputStream eagerly if there is no dictionary or - // if dictionary encoding has fallen back to PLAIN - if (has_dictionary_ && !fallback_) { // Save pages until end of dictionary encoding - std::shared_ptr<Buffer> compressed_data_copy; - PARQUET_THROW_NOT_OK(compressed_data->Copy( - 0, compressed_data->size(), allocator_, &compressed_data_copy)); - CompressedDataPage page(compressed_data_copy, - static_cast<int32_t>(num_buffered_values_), encoding_, Encoding::RLE, - Encoding::RLE, uncompressed_size, page_stats); - data_pages_.push_back(std::move(page)); - } else { // Eagerly write pages - CompressedDataPage page(compressed_data, static_cast<int32_t>(num_buffered_values_), - encoding_, Encoding::RLE, Encoding::RLE, uncompressed_size, page_stats); - WriteDataPage(page); - } - - // Re-initialize the sinks for next Page. - InitSinks(); - num_buffered_values_ = 0; - num_buffered_encoded_values_ = 0; -} - -void ColumnWriter::WriteDataPage(const CompressedDataPage& page) { - total_bytes_written_ += pager_->WriteDataPage(page); -} - -int64_t ColumnWriter::Close() { - if (!closed_) { - closed_ = true; - if (has_dictionary_ && !fallback_) { WriteDictionaryPage(); } - - FlushBufferedDataPages(); - - EncodedStatistics chunk_statistics = GetChunkStatistics(); - if (chunk_statistics.is_set()) metadata_->SetStatistics(chunk_statistics); - pager_->Close(has_dictionary_, fallback_); - } - - if (num_rows_ != expected_rows_) { - std::stringstream ss; - ss << "Written rows: " << num_rows_ << " != expected rows: " << expected_rows_ - << "in the current column chunk"; - throw ParquetException(ss.str()); - } - - return total_bytes_written_; -} - -void ColumnWriter::FlushBufferedDataPages() { - // Write all outstanding data to a new page - if (num_buffered_values_ > 0) { AddDataPage(); } - for (size_t i = 0; i < data_pages_.size(); i++) { - WriteDataPage(data_pages_[i]); - } - data_pages_.clear(); -} - -// ---------------------------------------------------------------------- -// TypedColumnWriter - -template <typename Type> -TypedColumnWriter<Type>::TypedColumnWriter(ColumnChunkMetaDataBuilder* metadata, - std::unique_ptr<PageWriter> pager, int64_t expected_rows, Encoding::type encoding, - const WriterProperties* properties) - : ColumnWriter(metadata, std::move(pager), expected_rows, - (encoding == Encoding::PLAIN_DICTIONARY || - encoding == Encoding::RLE_DICTIONARY), - encoding, properties) { - switch (encoding) { - case Encoding::PLAIN: - current_encoder_.reset(new PlainEncoder<Type>(descr_, properties->memory_pool())); - break; - case Encoding::PLAIN_DICTIONARY: - case Encoding::RLE_DICTIONARY: - current_encoder_.reset( - new DictEncoder<Type>(descr_, &pool_, properties->memory_pool())); - break; - default: - ParquetException::NYI("Selected encoding is not supported"); - } - - if (properties->statistics_enabled(descr_->path())) { - page_statistics_ = std::unique_ptr<TypedStats>(new TypedStats(descr_, allocator_)); - chunk_statistics_ = std::unique_ptr<TypedStats>(new TypedStats(descr_, allocator_)); - } -} - -// Only one Dictionary Page is written. -// Fallback to PLAIN if dictionary page limit is reached. -template <typename Type> -void TypedColumnWriter<Type>::CheckDictionarySizeLimit() { - auto dict_encoder = static_cast<DictEncoder<Type>*>(current_encoder_.get()); - if (dict_encoder->dict_encoded_size() >= properties_->dictionary_pagesize_limit()) { - WriteDictionaryPage(); - // Serialize the buffered Dictionary Indicies - FlushBufferedDataPages(); - fallback_ = true; - // Only PLAIN encoding is supported for fallback in V1 - current_encoder_.reset(new PlainEncoder<Type>(descr_, properties_->memory_pool())); - encoding_ = Encoding::PLAIN; - } -} - -template <typename Type> -void TypedColumnWriter<Type>::WriteDictionaryPage() { - auto dict_encoder = static_cast<DictEncoder<Type>*>(current_encoder_.get()); - std::shared_ptr<PoolBuffer> buffer = - AllocateBuffer(properties_->memory_pool(), dict_encoder->dict_encoded_size()); - dict_encoder->WriteDict(buffer->mutable_data()); - // TODO Get rid of this deep call - dict_encoder->mem_pool()->FreeAll(); - - DictionaryPage page( - buffer, dict_encoder->num_entries(), properties_->dictionary_index_encoding()); - total_bytes_written_ += pager_->WriteDictionaryPage(page); -} - -template <typename Type> -EncodedStatistics TypedColumnWriter<Type>::GetPageStatistics() { - EncodedStatistics result; - if (page_statistics_) result = page_statistics_->Encode(); - return result; -} - -template <typename Type> -EncodedStatistics TypedColumnWriter<Type>::GetChunkStatistics() { - EncodedStatistics result; - if (chunk_statistics_) result = chunk_statistics_->Encode(); - return result; -} - -template <typename Type> -void TypedColumnWriter<Type>::ResetPageStatistics() { - if (chunk_statistics_ != nullptr) { - chunk_statistics_->Merge(*page_statistics_); - page_statistics_->Reset(); - } -} - -// ---------------------------------------------------------------------- -// Dynamic column writer constructor - -std::shared_ptr<ColumnWriter> ColumnWriter::Make(ColumnChunkMetaDataBuilder* metadata, - std::unique_ptr<PageWriter> pager, int64_t expected_rows, - const WriterProperties* properties) { - const ColumnDescriptor* descr = metadata->descr(); - Encoding::type encoding = properties->encoding(descr->path()); - if (properties->dictionary_enabled(descr->path()) && - descr->physical_type() != Type::BOOLEAN) { - encoding = properties->dictionary_page_encoding(); - } - switch (descr->physical_type()) { - case Type::BOOLEAN: - return std::make_shared<BoolWriter>( - metadata, std::move(pager), expected_rows, encoding, properties); - case Type::INT32: - return std::make_shared<Int32Writer>( - metadata, std::move(pager), expected_rows, encoding, properties); - case Type::INT64: - return std::make_shared<Int64Writer>( - metadata, std::move(pager), expected_rows, encoding, properties); - case Type::INT96: - return std::make_shared<Int96Writer>( - metadata, std::move(pager), expected_rows, encoding, properties); - case Type::FLOAT: - return std::make_shared<FloatWriter>( - metadata, std::move(pager), expected_rows, encoding, properties); - case Type::DOUBLE: - return std::make_shared<DoubleWriter>( - metadata, std::move(pager), expected_rows, encoding, properties); - case Type::BYTE_ARRAY: - return std::make_shared<ByteArrayWriter>( - metadata, std::move(pager), expected_rows, encoding, properties); - case Type::FIXED_LEN_BYTE_ARRAY: - return std::make_shared<FixedLenByteArrayWriter>( - metadata, std::move(pager), expected_rows, encoding, properties); - default: - ParquetException::NYI("type reader not implemented"); - } - // Unreachable code, but supress compiler warning - return std::shared_ptr<ColumnWriter>(nullptr); -} - -// ---------------------------------------------------------------------- -// Instantiate templated classes - -template <typename DType> -inline int64_t TypedColumnWriter<DType>::WriteMiniBatch(int64_t num_values, - const int16_t* def_levels, const int16_t* rep_levels, const 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; } - } - - WriteDefinitionLevels(num_values, def_levels); - } else { - // Required field, write all values - values_to_write = num_values; - } - - // Not present for non-repeated fields - if (descr_->max_repetition_level() > 0) { - // 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_++; } - } - - WriteRepetitionLevels(num_values, rep_levels); - } else { - // Each value is exactly one row - num_rows_ += static_cast<int>(num_values); - } - - if (num_rows_ > expected_rows_) { - throw ParquetException("More rows were written in the column chunk than expected"); - } - - // PARQUET-780 - if (values_to_write > 0) { DCHECK(nullptr != values) << "Values ptr cannot be NULL"; } - - WriteValues(values_to_write, values); - - if (page_statistics_ != nullptr) { - page_statistics_->Update(values, values_to_write, num_values - values_to_write); - } - - num_buffered_values_ += num_values; - num_buffered_encoded_values_ += values_to_write; - - if (current_encoder_->EstimatedDataEncodedSize() >= properties_->data_pagesize()) { - AddDataPage(); - } - if (has_dictionary_ && !fallback_) { CheckDictionarySizeLimit(); } - - return values_to_write; -} - -template <typename DType> -inline int64_t TypedColumnWriter<DType>::WriteMiniBatchSpaced(int64_t num_values, - const int16_t* def_levels, const int16_t* rep_levels, const uint8_t* valid_bits, - int64_t valid_bits_offset, const T* values, int64_t* num_spaced_written) { - int64_t values_to_write = 0; - int64_t spaced_values_to_write = 0; - // If the field is required and non-repeated, there are no definition levels - if (descr_->max_definition_level() > 0) { - // Minimal definition level for which spaced values are written - int16_t min_spaced_def_level = descr_->max_definition_level(); - if (descr_->schema_node()->is_optional()) { min_spaced_def_level--; } - for (int64_t i = 0; i < num_values; ++i) { - if (def_levels[i] == descr_->max_definition_level()) { ++values_to_write; } - if (def_levels[i] >= min_spaced_def_level) { ++spaced_values_to_write; } - } - - WriteDefinitionLevels(num_values, def_levels); - } else { - // Required field, write all values - values_to_write = num_values; - spaced_values_to_write = num_values; - } - - // Not present for non-repeated fields - if (descr_->max_repetition_level() > 0) { - // 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_++; } - } - - WriteRepetitionLevels(num_values, rep_levels); - } else { - // Each value is exactly one row - num_rows_ += static_cast<int>(num_values); - } - - if (num_rows_ > expected_rows_) { - throw ParquetException("More rows were written in the column chunk than expected"); - } - - if (descr_->schema_node()->is_optional()) { - WriteValuesSpaced(spaced_values_to_write, valid_bits, valid_bits_offset, values); - } else { - WriteValues(values_to_write, values); - } - *num_spaced_written = spaced_values_to_write; - - if (page_statistics_ != nullptr) { - page_statistics_->UpdateSpaced(values, valid_bits, valid_bits_offset, values_to_write, - num_values - values_to_write); - } - - num_buffered_values_ += num_values; - num_buffered_encoded_values_ += values_to_write; - - if (current_encoder_->EstimatedDataEncodedSize() >= properties_->data_pagesize()) { - AddDataPage(); - } - if (has_dictionary_ && !fallback_) { CheckDictionarySizeLimit(); } - - return values_to_write; -} - -template <typename DType> -void TypedColumnWriter<DType>::WriteBatch(int64_t num_values, const int16_t* def_levels, - const int16_t* rep_levels, const T* values) { - // We check for DataPage limits only after we have inserted the values. If a user - // writes a large number of values, the DataPage size can be much above the limit. - // The purpose of this chunking is to bound this. Even if a user writes large number - // of values, the chunking will ensure the AddDataPage() is called at a reasonable - // pagesize limit - int64_t write_batch_size = properties_->write_batch_size(); - int num_batches = static_cast<int>(num_values / write_batch_size); - int64_t num_remaining = num_values % write_batch_size; - int64_t value_offset = 0; - for (int round = 0; round < num_batches; round++) { - int64_t offset = round * write_batch_size; - int64_t num_values = WriteMiniBatch(write_batch_size, &def_levels[offset], - &rep_levels[offset], &values[value_offset]); - value_offset += num_values; - } - // Write the remaining values - int64_t offset = num_batches * write_batch_size; - WriteMiniBatch( - num_remaining, &def_levels[offset], &rep_levels[offset], &values[value_offset]); -} - -template <typename DType> -void TypedColumnWriter<DType>::WriteBatchSpaced(int64_t num_values, - const int16_t* def_levels, const int16_t* rep_levels, const uint8_t* valid_bits, - int64_t valid_bits_offset, const T* values) { - // We check for DataPage limits only after we have inserted the values. If a user - // writes a large number of values, the DataPage size can be much above the limit. - // The purpose of this chunking is to bound this. Even if a user writes large number - // of values, the chunking will ensure the AddDataPage() is called at a reasonable - // pagesize limit - int64_t write_batch_size = properties_->write_batch_size(); - int num_batches = static_cast<int>(num_values / write_batch_size); - int64_t num_remaining = num_values % write_batch_size; - int64_t num_spaced_written = 0; - int64_t values_offset = 0; - for (int round = 0; round < num_batches; round++) { - int64_t offset = round * write_batch_size; - WriteMiniBatchSpaced(write_batch_size, &def_levels[offset], &rep_levels[offset], - valid_bits, valid_bits_offset + values_offset, values + values_offset, - &num_spaced_written); - values_offset += num_spaced_written; - } - // Write the remaining values - int64_t offset = num_batches * write_batch_size; - WriteMiniBatchSpaced(num_remaining, &def_levels[offset], &rep_levels[offset], - valid_bits, valid_bits_offset + values_offset, values + values_offset, - &num_spaced_written); -} - -template <typename DType> -void TypedColumnWriter<DType>::WriteValues(int64_t num_values, const T* values) { - current_encoder_->Put(values, static_cast<int>(num_values)); -} - -template <typename DType> -void TypedColumnWriter<DType>::WriteValuesSpaced(int64_t num_values, - const uint8_t* valid_bits, int64_t valid_bits_offset, const T* values) { - current_encoder_->PutSpaced( - values, static_cast<int>(num_values), valid_bits, valid_bits_offset); -} - -template class PARQUET_TEMPLATE_EXPORT TypedColumnWriter<BooleanType>; -template class PARQUET_TEMPLATE_EXPORT TypedColumnWriter<Int32Type>; -template class PARQUET_TEMPLATE_EXPORT TypedColumnWriter<Int64Type>; -template class PARQUET_TEMPLATE_EXPORT TypedColumnWriter<Int96Type>; -template class PARQUET_TEMPLATE_EXPORT TypedColumnWriter<FloatType>; -template class PARQUET_TEMPLATE_EXPORT TypedColumnWriter<DoubleType>; -template class PARQUET_TEMPLATE_EXPORT TypedColumnWriter<ByteArrayType>; -template class PARQUET_TEMPLATE_EXPORT TypedColumnWriter<FLBAType>; - -} // namespace parquet
