enhance: make parquet open_async non-blocking asynchronous - #600
Conversation
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: jiaqizho The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #600 +/- ##
==========================================
- Coverage 76.41% 76.37% -0.05%
==========================================
Files 173 173
Lines 17561 17749 +188
Branches 2655 2681 +26
==========================================
+ Hits 13419 13555 +136
- Misses 4142 4194 +52
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
bf9448e to
2d8b7a9
Compare
| return; | ||
| } | ||
|
|
||
| status = state->reader->finish_open(std::shared_ptr<::parquet::arrow::FileReader>(std::move(arrow_reader))); |
There was a problem hiding this comment.
finish_open() is called inside the OpenAsync completion callback here with no surrounding try/catch, but it can throw: finish_open -> create_row_group_infos -> try_build_row_group_infos -> RowGroupMetadata::Deserialize throws std::runtime_error / std::stoull on malformed row-group metadata (metadata.cpp:203,206). A thrown exception escapes the arrow Future callback instead of being converted to arrow::Status, so state->promise is never set and the caller observes a folly BrokenPromise (or, when driven inline, the raw C++ exception) rather than a cleanly failed future. Wrap the finalization body in try/catch and convert thrown exceptions to arrow::Status before calling complete().
| fmt::format("Parquet file size is negative. [path={}, file_size={}]", state->reader->path_, file_size))); | ||
| } | ||
| state->file_size = file_size; | ||
| state->reader->file_size_ = static_cast<uint64_t>(file_size); |
There was a problem hiding this comment.
The file size resolved by the async HEAD is written only to the transient reader's file_size_ here; MetaTrait::Payload has no size field and create_metadata_from_reader never stores it, so the resolved size never reaches the cached Metadata. When the caller-supplied kPropertyFileSize is 0 (the same case that required the async HEAD), cache-hit reconstruction via create_from_metadata opens with file_size=0, and Parquet's SerializedFile constructor calls source_->GetSize() unconditionally before the metadata check (parquet/file_reader.cc:304), forcing ObjectCrtInputFile::GetSize -> EnsureHeadObject to issue a blocking synchronous S3 HEAD even though the footer metadata is cached. Persist the resolved size (e.g. add it to Payload and prefer it in create_from_metadata) so reconstruction reuses it.
2d8b7a9 to
1fa45cb
Compare
| metadata = try_parse_footer_metadata(footer_buffer, reader_props); | ||
| } | ||
|
|
||
| ARROW_RETURN_NOT_OK(builder.Open(std::move(parquet_file), reader_props, metadata)); |
There was a problem hiding this comment.
The cache-backed reconstruction path falls back to a synchronous builder.Open() that performs blocking footer I/O, and because the guard at line 363 is if (key_retriever) (retriever-existence, not footer encryption) any retriever-configured reader — plaintext included — takes this path, not just encrypted footers. This blocks the async executor thread on a network footer read, defeating the async-open contract, and is the limitation flagged by the existing FIXME. Reconstruct and open the footer asynchronously so retriever-configured readers don't stall the caller.
| } | ||
|
|
||
| ctx->self->SetCachedContentLength(content_length); | ||
| ctx->future.MarkFinished(content_length); |
There was a problem hiding this comment.
When open_async's future is consumed with an inline executor (e.g. .via(&folly::InlineExecutor::instance())), FollyArrowExecutor forwards each TransferAlways continuation straight into executor_->add, so the next step runs inline inside the AWS callback that publishes the previous result. GetSizeAsync's HEAD callback calls ctx->future.MarkFinished (s3_filesystem.cpp:838) while AsyncHeadContext still holds the finalizer std::shared_lock, and the chained footer read then calls holder_->Lock() again (s3_filesystem.cpp:870), recursively acquiring a second shared_lock on the same finalizer mutex on one thread — undefined behavior, and a deadlock against a concurrent FinalizeS3() writer on a writer-preferring shared_mutex. The new footer-to-missing-metadata read pair self-chains the same way (both reads go through ReadAtAsyncInto, which also holds its lock across MarkFinished), so the result must be published after the client/finalizer guard is released in both GetSizeAsync and the read path, not only for the HEAD-to-GET hop.
jiaqizho
changed the title
Previously, open_async only moved the synchronous Parquet open path onto a Folly executor. Opening a remote file could still block a worker while fetching the file size and footer, so the async API did not provide a fully non-blocking open flow. The new path uses ParquetFileReader::OpenAsync and bridges the Arrow futures back to the Folly result. When a usable footer-size hint is available, it reads that suffix directly, parses the trailer and metadata, and fetches only the missing metadata range when the hint is too small. Encrypted files and unusable hints fallback to Parquet’s native asynchronous footer handling. The non-blocking random-access interface now also supports asynchronous size lookup, backed by S3 CRT HeadObjectAsync with cached size reuse and existing error mapping. Open-time continuations are transferred onto the executor supplied through via(), while normal data reads return to their original I/O path after the reader has finished opening. Signed-off-by: jiaqizho <jiaqi.zhou@zilliz.com>
1fa45cb to
4bc3068
Compare
3 participants
Previously, open_async only moved the synchronous Parquet open path onto a Folly executor. Opening a remote file could still block a worker while fetching the file size and footer, so the async API did not provide a fully non-blocking open flow.
The new path uses ParquetFileReader::OpenAsync and bridges the Arrow futures back to the Folly result. When a usable footer-size hint is available, it reads that suffix directly, parses the trailer and metadata, and fetches only the missing metadata range when the hint is too small. Encrypted files and unusable hints fallback to Parquet’s native asynchronous footer handling.
The non-blocking random-access interface now also supports asynchronous size lookup, backed by S3 CRT HeadObjectAsync with cached size reuse and existing error mapping. Open-time continuations are transferred onto the executor supplied through via(), while normal data reads return to their original I/O path after the reader has finished opening.