-
Notifications
You must be signed in to change notification settings - Fork 4.2k
GH-41579: [C++][Python][Parquet] Support reading/writing key-value metadata from/to ColumnChunkMetaData #41580
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 21 commits
Commits
Show all changes
22 commits
Select commit
Hold shift + click to select a range
9aeb8b1
Support reading and writing column chunk key-value metadata
clee704 408270a
Python support
clee704 008d6cb
Move the test file to parquet-testing
clee704 d23647e
Address review comments
clee704 db6b50d
Update Python test
clee704 d920564
Address review comment
clee704 5a5c242
Address review comment
clee704 4379c74
Address review comment
clee704 eb5cfa2
Revert accidentally committed change
clee704 3179912
Add test case for ResetKeyValueMetadata
clee704 197ef0f
Remove using
clee704 de099d8
Address review comment
clee704 cd5130e
Update test
clee704 1288d8e
Merge branch 'main' into columnchunkkeyvaluemetadata
mapleFU ef4ae5b
fix cpp comments
mapleFU 807b01f
trying to fix python lint
mapleFU e5f0b9d
update submodules
mapleFU 1faf669
update some non-move impl
mapleFU 25f2a25
Trying to fix python tests
mapleFU c54926c
Merge branch 'main' into columnchunkkeyvaluemetadata
mapleFU 96fc780
Merge branch 'main' into columnchunkkeyvaluemetadata
mapleFU ab5cb56
Apply suggestions
mapleFU File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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 |
|---|---|---|
|
|
@@ -23,10 +23,12 @@ | |
| #include <gtest/gtest.h> | ||
|
|
||
| #include "arrow/io/buffered.h" | ||
| #include "arrow/io/file.h" | ||
| #include "arrow/testing/gtest_util.h" | ||
| #include "arrow/util/bit_util.h" | ||
| #include "arrow/util/bitmap_builders.h" | ||
| #include "arrow/util/config.h" | ||
| #include "arrow/util/key_value_metadata.h" | ||
|
|
||
| #include "parquet/column_page.h" | ||
| #include "parquet/column_reader.h" | ||
|
|
@@ -51,6 +53,9 @@ using schema::PrimitiveNode; | |
|
|
||
| namespace test { | ||
|
|
||
| using ::testing::IsNull; | ||
| using ::testing::NotNull; | ||
|
|
||
| // The default size used in most tests. | ||
| const int SMALL_SIZE = 100; | ||
| #ifdef PARQUET_VALGRIND | ||
|
|
@@ -385,6 +390,15 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> { | |
| return metadata_accessor->encoding_stats(); | ||
| } | ||
|
|
||
| std::shared_ptr<const KeyValueMetadata> metadata_key_value_metadata() { | ||
| // Metadata accessor must be created lazily. | ||
| // This is because the ColumnChunkMetaData semantics dictate the metadata object is | ||
| // complete (no changes to the metadata buffer can be made after instantiation) | ||
| auto metadata_accessor = | ||
| ColumnChunkMetaData::Make(metadata_->contents(), this->descr_); | ||
| return metadata_accessor->key_value_metadata(); | ||
| } | ||
|
|
||
| protected: | ||
| int64_t values_read_; | ||
| // Keep the reader alive as for ByteArray the lifetime of the ByteArray | ||
|
|
@@ -1705,5 +1719,55 @@ TEST(TestColumnWriter, WriteDataPageV2HeaderNullCount) { | |
| } | ||
| } | ||
|
|
||
| using TestInt32Writer = TestPrimitiveWriter<Int32Type>; | ||
|
|
||
| TEST_F(TestInt32Writer, NoWriteKeyValueMetadata) { | ||
| auto writer = this->BuildWriter(); | ||
| writer->Close(); | ||
| auto key_value_metadata = metadata_key_value_metadata(); | ||
| ASSERT_THAT(key_value_metadata, IsNull()); | ||
| } | ||
|
|
||
| TEST_F(TestInt32Writer, WriteKeyValueMetadata) { | ||
| auto writer = this->BuildWriter(); | ||
| writer->AddKeyValueMetadata(KeyValueMetadata::Make({"hello"}, {"world"})); | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can you add a test where |
||
| writer->Close(); | ||
| auto key_value_metadata = metadata_key_value_metadata(); | ||
| ASSERT_THAT(key_value_metadata, NotNull()); | ||
| ASSERT_EQ(1, key_value_metadata->size()); | ||
| ASSERT_OK_AND_ASSIGN(auto value, key_value_metadata->Get("hello")); | ||
| ASSERT_EQ("world", value); | ||
| } | ||
|
|
||
| TEST_F(TestInt32Writer, ResetKeyValueMetadata) { | ||
| auto writer = this->BuildWriter(); | ||
| writer->AddKeyValueMetadata(KeyValueMetadata::Make({"hello"}, {"world"})); | ||
| writer->ResetKeyValueMetadata(); | ||
| writer->Close(); | ||
| auto key_value_metadata = metadata_key_value_metadata(); | ||
| ASSERT_THAT(key_value_metadata, IsNull()); | ||
| } | ||
|
|
||
| TEST_F(TestInt32Writer, WriteKeyValueMetadataEndToEnd) { | ||
| auto sink = CreateOutputStream(); | ||
| { | ||
| auto file_writer = ParquetFileWriter::Open( | ||
| sink, std::dynamic_pointer_cast<schema::GroupNode>(schema_.schema_root())); | ||
| auto rg_writer = file_writer->AppendRowGroup(); | ||
| auto col_writer = rg_writer->NextColumn(); | ||
| col_writer->AddKeyValueMetadata(KeyValueMetadata::Make({"foo"}, {"bar"})); | ||
| file_writer->Close(); | ||
| } | ||
| ASSERT_OK_AND_ASSIGN(auto buffer, sink->Finish()); | ||
| auto file_reader = | ||
| ParquetFileReader::Open(std::make_shared<::arrow::io::BufferReader>(buffer)); | ||
| auto key_value_metadata = | ||
| file_reader->metadata()->RowGroup(0)->ColumnChunk(0)->key_value_metadata(); | ||
| ASSERT_THAT(key_value_metadata, NotNull()); | ||
| ASSERT_EQ(1U, key_value_metadata->size()); | ||
| ASSERT_OK_AND_ASSIGN(auto value, key_value_metadata->Get("foo")); | ||
| ASSERT_EQ("bar", value); | ||
| } | ||
|
|
||
| } // namespace test | ||
| } // namespace parquet | ||
This file contains hidden or 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 |
|---|---|---|
|
|
@@ -135,6 +135,36 @@ std::shared_ptr<Statistics> MakeColumnStats(const format::ColumnMetaData& meta_d | |
| throw ParquetException("Can't decode page statistics for selected column type"); | ||
| } | ||
|
|
||
| template <typename Metadata> | ||
| std::shared_ptr<KeyValueMetadata> CopyKeyValueMetadata(const Metadata& source) { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Perhaps call this |
||
| std::shared_ptr<KeyValueMetadata> metadata = nullptr; | ||
| if (source.__isset.key_value_metadata) { | ||
| std::vector<std::string> keys; | ||
| std::vector<std::string> values; | ||
| keys.reserve(source.key_value_metadata.size()); | ||
| values.reserve(source.key_value_metadata.size()); | ||
| for (const auto& it : source.key_value_metadata) { | ||
| keys.push_back(it.key); | ||
| values.push_back(it.value); | ||
| } | ||
| metadata = std::make_shared<KeyValueMetadata>(std::move(keys), std::move(values)); | ||
| } | ||
| return metadata; | ||
| } | ||
|
|
||
| template <typename Metadata> | ||
| void ToThriftKeyValueMetadata(const KeyValueMetadata& source, Metadata* metadata) { | ||
| std::vector<format::KeyValue> key_value_metadata; | ||
| key_value_metadata.reserve(static_cast<size_t>(source.size())); | ||
| for (int64_t i = 0; i < source.size(); ++i) { | ||
| format::KeyValue kv_pair; | ||
| kv_pair.__set_key(source.key(i)); | ||
| kv_pair.__set_value(source.value(i)); | ||
| key_value_metadata.emplace_back(std::move(kv_pair)); | ||
| } | ||
| metadata->__set_key_value_metadata(std::move(key_value_metadata)); | ||
| } | ||
|
|
||
| // MetaData Accessor | ||
|
|
||
| // ColumnCryptoMetaData | ||
|
|
@@ -233,6 +263,7 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl { | |
| encoding_stats.count}); | ||
| } | ||
| possible_stats_ = nullptr; | ||
| InitKeyValueMetadata(); | ||
| } | ||
|
|
||
| bool Equals(const ColumnChunkMetaDataImpl& other) const { | ||
|
|
@@ -343,7 +374,15 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl { | |
| return std::nullopt; | ||
| } | ||
|
|
||
| const std::shared_ptr<const KeyValueMetadata>& key_value_metadata() const { | ||
| return key_value_metadata_; | ||
| } | ||
|
|
||
| private: | ||
| void InitKeyValueMetadata() { | ||
| key_value_metadata_ = CopyKeyValueMetadata(*column_metadata_); | ||
| } | ||
|
|
||
| mutable std::shared_ptr<Statistics> possible_stats_; | ||
| std::vector<Encoding::type> encodings_; | ||
| std::vector<PageEncodingStats> encoding_stats_; | ||
|
|
@@ -353,6 +392,7 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl { | |
| const ColumnDescriptor* descr_; | ||
| const ReaderProperties properties_; | ||
| const ApplicationVersion* writer_version_; | ||
| std::shared_ptr<const KeyValueMetadata> key_value_metadata_; | ||
| }; | ||
|
|
||
| std::unique_ptr<ColumnChunkMetaData> ColumnChunkMetaData::Make( | ||
|
|
@@ -471,6 +511,11 @@ bool ColumnChunkMetaData::Equals(const ColumnChunkMetaData& other) const { | |
| return impl_->Equals(*other.impl_); | ||
| } | ||
|
|
||
| const std::shared_ptr<const KeyValueMetadata>& ColumnChunkMetaData::key_value_metadata() | ||
| const { | ||
| return impl_->key_value_metadata(); | ||
| } | ||
|
|
||
| // row-group metadata | ||
| class RowGroupMetaData::RowGroupMetaDataImpl { | ||
| public: | ||
|
|
@@ -913,7 +958,7 @@ class FileMetaData::FileMetaDataImpl { | |
| std::vector<parquet::ColumnOrder> column_orders; | ||
| if (metadata_->__isset.column_orders) { | ||
| column_orders.reserve(metadata_->column_orders.size()); | ||
| for (auto column_order : metadata_->column_orders) { | ||
| for (auto& column_order : metadata_->column_orders) { | ||
| if (column_order.__isset.TYPE_ORDER) { | ||
| column_orders.push_back(ColumnOrder::type_defined_); | ||
| } else { | ||
|
|
@@ -927,16 +972,7 @@ class FileMetaData::FileMetaDataImpl { | |
| schema_.updateColumnOrders(column_orders); | ||
| } | ||
|
|
||
| void InitKeyValueMetadata() { | ||
| std::shared_ptr<KeyValueMetadata> metadata = nullptr; | ||
| if (metadata_->__isset.key_value_metadata) { | ||
| metadata = std::make_shared<KeyValueMetadata>(); | ||
| for (const auto& it : metadata_->key_value_metadata) { | ||
| metadata->Append(it.key, it.value); | ||
| } | ||
| } | ||
| key_value_metadata_ = std::move(metadata); | ||
| } | ||
| void InitKeyValueMetadata() { key_value_metadata_ = CopyKeyValueMetadata(*metadata_); } | ||
| }; | ||
|
|
||
| std::shared_ptr<FileMetaData> FileMetaData::Make( | ||
|
|
@@ -1590,6 +1626,10 @@ class ColumnChunkMetaDataBuilder::ColumnChunkMetaDataBuilderImpl { | |
| column_chunk_->meta_data.__set_encodings(std::move(thrift_encodings)); | ||
| column_chunk_->meta_data.__set_encoding_stats(std::move(thrift_encoding_stats)); | ||
|
|
||
| if (key_value_metadata_) { | ||
| ToThriftKeyValueMetadata(*key_value_metadata_, &column_chunk_->meta_data); | ||
| } | ||
|
|
||
| const auto& encrypt_md = | ||
| properties_->column_encryption_properties(column_->path()->ToDotString()); | ||
| // column is encrypted | ||
|
|
@@ -1656,6 +1696,10 @@ class ColumnChunkMetaDataBuilder::ColumnChunkMetaDataBuilderImpl { | |
| return column_chunk_->meta_data.total_compressed_size; | ||
| } | ||
|
|
||
| void SetKeyValueMetadata(std::shared_ptr<const KeyValueMetadata> key_value_metadata) { | ||
| key_value_metadata_ = std::move(key_value_metadata); | ||
| } | ||
|
|
||
| private: | ||
| void Init(format::ColumnChunk* column_chunk) { | ||
| column_chunk_ = column_chunk; | ||
|
|
@@ -1670,6 +1714,7 @@ class ColumnChunkMetaDataBuilder::ColumnChunkMetaDataBuilderImpl { | |
| std::unique_ptr<format::ColumnChunk> owned_column_chunk_; | ||
| const std::shared_ptr<WriterProperties> properties_; | ||
| const ColumnDescriptor* column_; | ||
| std::shared_ptr<const KeyValueMetadata> key_value_metadata_; | ||
| }; | ||
|
|
||
| std::unique_ptr<ColumnChunkMetaDataBuilder> ColumnChunkMetaDataBuilder::Make( | ||
|
|
@@ -1727,6 +1772,11 @@ void ColumnChunkMetaDataBuilder::SetStatistics(const EncodedStatistics& result) | |
| impl_->SetStatistics(result); | ||
| } | ||
|
|
||
| void ColumnChunkMetaDataBuilder::SetKeyValueMetadata( | ||
| std::shared_ptr<const KeyValueMetadata> key_value_metadata) { | ||
| impl_->SetKeyValueMetadata(std::move(key_value_metadata)); | ||
| } | ||
|
|
||
| int64_t ColumnChunkMetaDataBuilder::total_compressed_size() const { | ||
| return impl_->total_compressed_size(); | ||
| } | ||
|
|
@@ -1925,16 +1975,7 @@ class FileMetaDataBuilder::FileMetaDataBuilderImpl { | |
| } else if (key_value_metadata) { | ||
| key_value_metadata_ = key_value_metadata_->Merge(*key_value_metadata); | ||
| } | ||
| metadata_->key_value_metadata.clear(); | ||
| metadata_->key_value_metadata.reserve( | ||
| static_cast<size_t>(key_value_metadata_->size())); | ||
| for (int64_t i = 0; i < key_value_metadata_->size(); ++i) { | ||
| format::KeyValue kv_pair; | ||
| kv_pair.__set_key(key_value_metadata_->key(i)); | ||
| kv_pair.__set_value(key_value_metadata_->value(i)); | ||
| metadata_->key_value_metadata.push_back(std::move(kv_pair)); | ||
| } | ||
| metadata_->__isset.key_value_metadata = true; | ||
| ToThriftKeyValueMetadata(*key_value_metadata_, metadata_.get()); | ||
| } | ||
|
|
||
| int32_t file_version = 0; | ||
|
|
||
This file contains hidden or 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
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.