[C++][Parquet] GH-47628: Implement basic parquet file rewriter #47775
Conversation
|
Thanks for opening a pull request! If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project. Then could you also rename the pull request title in the following format? or See also: |
github-actions
bot
added
Component: Parquet
Component: C++
awaiting review
HuaHuaY
changed the title
HuaHuaY
force-pushed
the
fix_issue_47664
branch
from
e4de469 to
c216849
Compare
|
@pitrou @adamreeve @mapleFU Do you have any suggestions about this draft? Is there any efficient way to merge two parquet files' schema? |
| template <typename Builder> | ||
| void SerializeIndex( | ||
| const std::vector<std::vector<std::unique_ptr<Builder>>>& page_index_builders, | ||
| const std::vector<std::vector<std::unique_ptr<Index<Builder>>>>& page_indices, |
There was a problem hiding this comment.
Can this separate to different method? This reuse is a bit hacking to me
github-actions
bot
added
awaiting committer review
| /// \brief List of repetition level histograms for each page concatenated together. | ||
| virtual const std::vector<int64_t>& repetition_level_histograms() const = 0; | ||
|
|
||
| virtual const void* to_thrift() const = 0; |
There was a problem hiding this comment.
I'm not sure if this is on the critical path. Is it simpler and cleaner to convert ColumnIndex and OffsetIndex to their thrift equivalents?
There was a problem hiding this comment.
I plan to remove format::OffsetIndex in OffsetIndexImpl in my first commit, and add a new method static inline format::PageLocation ToThrift(const PageLocation&) in thrift_internal.h to convert offset index to its thrift format. But there was a format::ColumnIndex in TypedColumnIndexImpl which is used to avoid copying some vectors.
@pitrou Do you think we should preserve the ColumnIndex::to_thrift method or write a similar static inline format::ColumnIndex ToThrift(const TypedColumnIndexImpl<>&)?
| RowGroupRewriter(std::shared_ptr<ArrowInputFile> source, | ||
| std::shared_ptr<ArrowOutputStream> sink, | ||
| const RewriterProperties* props, | ||
| std::shared_ptr<RowGroupReader> row_group_reader, |
There was a problem hiding this comment.
Perhaps introduce a RowGroupContext to hold all row group xxx readers?
|
|
||
| // TODO(HuaHuaY): copy column index and bloom filter instead of reading and | ||
| // writing it. | ||
| auto column_index = page_index_reader_->GetColumnIndex(i); |
There was a problem hiding this comment.
We need to check if any of these is nullptr since they are optional.
| auto column_index = page_index_reader_->GetColumnIndex(i); | ||
| auto offset_index = page_index_reader_->GetOffsetIndex(i); | ||
| // TODO(HuaHuaY): support bloom filter writer | ||
| [[maybe_unused]] auto bloom_filter = |
| } | ||
|
|
||
| total_bytes_written += metadata_->total_byte_size(); | ||
| } else { |
There was a problem hiding this comment.
Perhaps we need to split this into smaller but dedicated functions?
cpp/src/parquet/file_rewriter.cc
Outdated
| bool HasData() { | ||
| while (current_rewriter_index_ < rewriters_.size() && | ||
| !rewriters_[current_rewriter_index_]->HasData()) { | ||
| rewriters_[current_rewriter_index_]->Close(); |
There was a problem hiding this comment.
Add a debug log to show the progress?
cpp/src/parquet/file_rewriter.cc
Outdated
| } | ||
| } | ||
|
|
||
| const SchemaDescriptor* schema() const { return rewriters_[0]->schema(); } |
There was a problem hiding this comment.
Is this still valid once Close has been called on it?
There was a problem hiding this comment.
Yes. rewriters_[0] isn't set to nullptr after Close is called in the current code.
| size_t current_rewriter_index_{}; | ||
| }; | ||
|
|
||
| class ExtendRewriter { |
There was a problem hiding this comment.
Do you have any suggestion? I just want to distinguish it from ConcatRewriter. If you don't like this name, should we rename it to MergeRewriter?
HuaHuaY
force-pushed
the
fix_issue_47664
branch
from
e037be7 to
253f281
Compare
4 participants
This is a draft PR now. I follow Java's implementation but I think it is not a good enough design for C++. Because we must copy lots of code from file_writer.cc or file_reader.cc and it will be troublesome to maintain in the future. I prefer to implement some classes inheriting
XXXWriterorXXXReader. I'll think about how to refactor the code. If anyone has any good suggestions, please comment.Now I have written two kinds of tests. Test the horizontal splicing and vertical splicing of parquet files separately. But only horizontal splicing is implemented now because I don't find an efficient way to merge two parquet files' schema.
Rationale for this change
Allow to rewrite parquet files in binary data formats instead of reading, decoding all values and writing them.
What changes are included in this PR?
ParquetFileRewriterandRewriterProperties.to_thriftandSetXXXmethods to help me copy the metadata.CopyStreammethods to callmemcpybetweenArrowInputStreamandArrowOutputStream.RowGroupMetaDataBuilder::NextColumnChunk(std::unique_ptr<ColumnChunkMetaData> cc_metadata, int64_t shift)which allows to add column metadata without creatingColumnChunkMetaDataBuilder.Are these changes tested?
Yes
Are there any user-facing changes?
ReaderProperties::GetStreamis changed to a const method. Only the signature has been changed. Its original implementation allows it to be declared as a const method.