forked from apache/arrow
-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
PARQUET-505: Column reader should automatically handle large data pages
This PR implements 1) PARQUET-505: Column reader should automatically handle large data pages 2) Adds support for Serialization 3) Test case for Serialization and Deserialization 4) Test case for SerializedPageReader and PARQUET-505 Author: Deepak Majeti <[email protected]> Closes apache#44 from majetideepak/PARQUET-505 and squashes the following commits: 4f754ba [Deepak Majeti] changed type of page header size defaults 4345812 [Deepak Majeti] PARQUET-505: Column reader should automatically handle large data pages Change-Id: I1b28296cabc0e6e776271b82ed58e267483911d3
- Loading branch information
1 parent
d35efe2
commit 4e18915
Showing
8 changed files
with
284 additions
and
14 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,109 @@ | ||
// 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 <cstdlib> | ||
#include <iostream> | ||
#include <sstream> | ||
#include <string> | ||
|
||
#include <gtest/gtest.h> | ||
|
||
#include "parquet/types.h" | ||
#include "parquet/thrift/parquet_types.h" | ||
#include "parquet/thrift/util.h" | ||
#include "parquet/column/serialized-page.h" | ||
#include "parquet/column/page.h" | ||
#include "parquet/column/reader.h" | ||
#include "parquet/column/test-util.h" | ||
|
||
|
||
namespace parquet_cpp { | ||
|
||
class TestSerializedPage : public ::testing::Test { | ||
public: | ||
void InitSerializedPageReader(const uint8_t* buffer, size_t header_size, | ||
parquet::CompressionCodec::type codec) { | ||
std::unique_ptr<InputStream> stream; | ||
stream.reset(new InMemoryInputStream(buffer, header_size)); | ||
page_reader_.reset(new SerializedPageReader(std::move(stream), codec)); | ||
} | ||
|
||
protected: | ||
std::unique_ptr<SerializedPageReader> page_reader_; | ||
}; | ||
|
||
TEST_F(TestSerializedPage, TestLargePageHeaders) { | ||
parquet::PageHeader in_page_header; | ||
parquet::DataPageHeader data_page_header; | ||
parquet::PageHeader out_page_header; | ||
parquet::Statistics stats; | ||
int expected_header_size = 512 * 1024; //512 KB | ||
int stats_size = 256 * 1024; // 256 KB | ||
std::string serialized_buffer; | ||
int num_values = 4141; | ||
|
||
InitStats(stats_size, stats); | ||
InitDataPage(stats, data_page_header, num_values); | ||
InitPageHeader(data_page_header, in_page_header); | ||
|
||
// Serialize the Page header | ||
ASSERT_NO_THROW(serialized_buffer = SerializeThriftMsg(&in_page_header, | ||
expected_header_size)); | ||
// check header size is between 256 KB to 16 MB | ||
ASSERT_LE(stats_size, serialized_buffer.length()); | ||
ASSERT_GE(DEFAULT_MAX_PAGE_HEADER_SIZE, serialized_buffer.length()); | ||
|
||
InitSerializedPageReader(reinterpret_cast<const uint8_t*>(serialized_buffer.c_str()), | ||
serialized_buffer.length(), parquet::CompressionCodec::UNCOMPRESSED); | ||
|
||
std::shared_ptr<Page> current_page = page_reader_->NextPage(); | ||
ASSERT_EQ(parquet::PageType::DATA_PAGE, current_page->type()); | ||
const DataPage* page = static_cast<const DataPage*>(current_page.get()); | ||
ASSERT_EQ(num_values, page->num_values()); | ||
} | ||
|
||
TEST_F(TestSerializedPage, TestFailLargePageHeaders) { | ||
parquet::PageHeader in_page_header; | ||
parquet::DataPageHeader data_page_header; | ||
parquet::PageHeader out_page_header; | ||
parquet::Statistics stats; | ||
int expected_header_size = 512 * 1024; // 512 KB | ||
int stats_size = 256 * 1024; // 256 KB | ||
int max_header_size = 128 * 1024; // 128 KB | ||
int num_values = 4141; | ||
std::string serialized_buffer; | ||
|
||
InitStats(stats_size, stats); | ||
InitDataPage(stats, data_page_header, num_values); | ||
InitPageHeader(data_page_header, in_page_header); | ||
|
||
// Serialize the Page header | ||
ASSERT_NO_THROW(serialized_buffer = SerializeThriftMsg(&in_page_header, | ||
expected_header_size)); | ||
// check header size is between 256 KB to 16 MB | ||
ASSERT_LE(stats_size, serialized_buffer.length()); | ||
ASSERT_GE(DEFAULT_MAX_PAGE_HEADER_SIZE, serialized_buffer.length()); | ||
|
||
InitSerializedPageReader(reinterpret_cast<const uint8_t*>(serialized_buffer.c_str()), | ||
serialized_buffer.length(), parquet::CompressionCodec::UNCOMPRESSED); | ||
|
||
// Set the max page header size to 128 KB, which is less than the current header size | ||
page_reader_->set_max_page_header_size(max_header_size); | ||
|
||
ASSERT_THROW(page_reader_->NextPage(), ParquetException); | ||
} | ||
} // namespace parquet_cpp |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -30,3 +30,5 @@ install(FILES | |
parquet_constants.h | ||
util.h | ||
DESTINATION include/parquet/thrift) | ||
|
||
ADD_PARQUET_TEST(serializer-test) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,78 @@ | ||
// 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 <cstdlib> | ||
#include <iostream> | ||
#include <sstream> | ||
#include <string> | ||
|
||
#include <gtest/gtest.h> | ||
|
||
#include "parquet/thrift/parquet_types.h" | ||
#include "parquet/thrift/util.h" | ||
#include "parquet/column/page.h" | ||
#include "parquet/column/reader.h" | ||
#include "parquet/column/test-util.h" | ||
|
||
using std::string; | ||
|
||
namespace parquet_cpp { | ||
|
||
class TestThrift : public ::testing::Test { | ||
|
||
}; | ||
|
||
TEST_F(TestThrift, TestSerializerDeserializer) { | ||
parquet::PageHeader in_page_header; | ||
parquet::DataPageHeader data_page_header; | ||
parquet::PageHeader out_page_header; | ||
parquet::Statistics stats; | ||
uint32_t max_header_len = 1024; | ||
uint32_t expected_header_size = 1024; | ||
uint32_t stats_size = 512; | ||
std::string serialized_buffer; | ||
int num_values = 4444; | ||
|
||
InitStats(stats_size, stats); | ||
InitDataPage(stats, data_page_header, num_values); | ||
InitPageHeader(data_page_header, in_page_header); | ||
|
||
// Serialize the Page header | ||
ASSERT_NO_THROW(serialized_buffer = SerializeThriftMsg(&in_page_header, expected_header_size)); | ||
ASSERT_LE(stats_size, serialized_buffer.length()); | ||
ASSERT_GE(max_header_len, serialized_buffer.length()); | ||
|
||
uint32_t header_size = 1024; | ||
// Deserialize the serialized page buffer | ||
ASSERT_NO_THROW(DeserializeThriftMsg(reinterpret_cast<const uint8_t*>(serialized_buffer.c_str()), | ||
&header_size, &out_page_header)); | ||
ASSERT_LE(stats_size, header_size); | ||
ASSERT_GE(max_header_len, header_size); | ||
|
||
ASSERT_EQ(parquet::Encoding::PLAIN, out_page_header.data_page_header.encoding); | ||
ASSERT_EQ(parquet::Encoding::RLE, out_page_header.data_page_header.definition_level_encoding); | ||
ASSERT_EQ(parquet::Encoding::RLE, out_page_header.data_page_header.repetition_level_encoding); | ||
for(int i = 0; i < stats_size; i++){ | ||
EXPECT_EQ(i % 255, (reinterpret_cast<const uint8_t*> | ||
(out_page_header.data_page_header.statistics.max.c_str()))[i]); | ||
} | ||
ASSERT_EQ(parquet::PageType::DATA_PAGE, out_page_header.type); | ||
ASSERT_EQ(num_values, out_page_header.data_page_header.num_values); | ||
|
||
} | ||
|
||
} // namespace parquet_cpp |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters