SteNicholas commented on code in PR #209:
URL: https://github.com/apache/paimon-cpp/pull/209#discussion_r3810802936
##########
src/paimon/common/utils/read_ahead_cache.cpp:
##########
@@ -89,38 +152,66 @@ class ReadAheadCache::Impl {
std::vector<std::atomic<bool>> is_cached_;
std::vector<ByteRange> pending_ranges_;
bool is_initialized_ = false;
+ // Statistics of the Read() requests issued to the cache, aggregated over
+ // all streams sharing this cache.
+ std::atomic<uint64_t> read_count_{0};
+ std::atomic<uint64_t> read_bytes_{0};
+ std::atomic<uint64_t> hits_{0};
+ std::atomic<uint64_t> hit_bytes_{0};
+ std::atomic<uint64_t> misses_{0};
+ std::atomic<uint64_t> miss_bytes_{0};
+ // Prefetch IO statistics: how many requests and bytes were actually issued
+ // to the underlying stream.
+ std::atomic<uint64_t> io_count_{0};
+ std::atomic<uint64_t> io_bytes_{0};
};
-void ReadAheadCache::Impl::Cache(std::vector<ByteRange> ranges) {
- std::sort(ranges.begin(), ranges.end(),
- [](const ByteRange& a, const ByteRange& b) { return a.offset <
b.offset; });
- std::vector<RangeCacheEntry> new_entries = MakeCacheEntries(ranges);
- // Add new entries, themselves ordered by offset
- std::unique_lock<std::shared_mutex> lock(rw_mutex_);
- if (entries_.size() > 0) {
- size_t new_entries_size = 0;
- for (const auto& e : new_entries) {
- new_entries_size += e.range.length;
- }
-
- size_t total_size = 0;
- for (const auto& e : entries_) {
- total_size += e.range.length;
+void ReadAheadCache::Impl::Cache(std::vector<size_t> pending_indices) {
+ std::vector<RangeCacheEntry> new_entries;
+ std::vector<PendingFetch> fetches;
+ // Mark is_cached_, publish the promise-backed entries and only then
+ // dispatch the IOs. The mark and the publication happen atomically under
+ // the write lock: a reader racing the prefetch observes is_cached_=true
+ // only once the covering entries are already visible, so it waits on
+ // their futures instead of issuing a duplicate underlying read.
+ {
+ std::unique_lock<std::shared_mutex> lock(rw_mutex_);
+ for (size_t idx : pending_indices) {
+ if (is_cached_[idx].exchange(true)) {
+ continue;
+ }
+ const ByteRange& range = pending_ranges_[idx];
+ auto promise = std::make_shared<std::promise<Status>>();
+ auto future = promise->get_future();
+ auto buffer = std::make_shared<Bytes>(range.length,
memory_pool_.get());
+ fetches.push_back({range, buffer, promise});
+ new_entries.emplace_back(range, std::move(buffer),
std::move(future));
}
- size_t limit = config_.GetBufferSizeLimit();
- while (!entries_.empty() && total_size + new_entries_size > limit) {
- auto iter = entries_.begin();
- total_size -= entries_.front().range.length;
- entries_.erase(iter);
+ if (!new_entries.empty()) {
+ // Add new entries, themselves ordered by offset
+ size_t new_entries_size = 0;
+ for (const auto& e : new_entries) {
+ new_entries_size += e.range.length;
+ }
+
+ size_t total_size = 0;
+ for (const auto& e : entries_) {
+ total_size += e.range.length;
+ }
+ size_t limit = config_.GetBufferSizeLimit();
+ while (!entries_.empty() && total_size + new_entries_size > limit)
{
+ auto iter = entries_.begin();
+ total_size -= entries_.front().range.length;
+ entries_.erase(iter);
Review Comment:
[P1] Keep evicted async reads tracked until completion
This can erase an entry whose asynchronous read is still in flight. The
callback keeps the buffer alive, but `ReleaseBuffers()` and the destructor only
wait for futures that remain in `entries_`; a later close/destruction can
therefore release the stream or memory pool while an evicted request is still
writing, and `Bytes` only retains a raw `MemoryPool*`. Please track every
dispatched request independently until completion and wait for all of them
during release, or avoid evicting in-flight entries.
##########
src/paimon/format/parquet/file_reader_wrapper.cpp:
##########
@@ -376,39 +393,68 @@ Status FileReaderWrapper::PrepareForReadingLazy(
target_row_groups_ = target_row_groups;
target_column_indices_ = column_indices;
reader_initialized_ = false;
+ pending_start_idx_.reset();
return Status::OK();
}
-std::vector<::arrow::io::ReadRange> FileReaderWrapper::CollectPreBufferRanges(
- const std::vector<int32_t>& column_indices) {
- std::vector<::arrow::io::ReadRange> ranges;
- auto file_metadata = file_reader_->parquet_reader()->metadata();
-
- for (const auto& trg : target_row_groups_) {
- if (trg.IsExcludedByReadRange()) continue;
-
- if (trg.IsPartiallyMatched()) {
- // Page-filtered RGs: only matching page byte ranges.
- auto row_group_page_index_reader =
GetRowGroupPageIndexReader(trg.GetRowGroupIndex());
- auto page_ranges = PageFilteredRowGroupReader::ComputePageRanges(
- trg, column_indices, row_group_page_index_reader,
file_reader_->parquet_reader());
- ranges.insert(ranges.end(),
std::make_move_iterator(page_ranges.begin()),
- std::make_move_iterator(page_ranges.end()));
- } else {
- // Fully-matched RGs: entire column chunk ranges.
- auto rg_metadata = file_metadata->RowGroup(trg.GetRowGroupIndex());
- for (int32_t col_idx : column_indices) {
- auto col_chunk = rg_metadata->ColumnChunk(col_idx);
- int64_t offset = col_chunk->data_page_offset();
- if (col_chunk->has_dictionary_page() &&
col_chunk->dictionary_page_offset() > 0 &&
- offset > col_chunk->dictionary_page_offset()) {
- offset = col_chunk->dictionary_page_offset();
+Result<std::vector<::arrow::io::ReadRange>>
FileReaderWrapper::CollectPreBufferRanges(
+ const std::vector<int32_t>& column_indices, uint64_t start_idx) {
+ return DoCollectPreBufferRanges(column_indices,
/*skip_read_range_excluded=*/true, start_idx);
+}
+
+Result<std::vector<::arrow::io::ReadRange>>
FileReaderWrapper::DoCollectPreBufferRanges(
+ const std::vector<int32_t>& column_indices, bool skip_read_range_excluded,
uint64_t start_idx) {
+ try {
+ std::vector<::arrow::io::ReadRange> ranges;
+ auto file_metadata = file_reader_->parquet_reader()->metadata();
+
+ for (uint64_t idx = start_idx; idx < target_row_groups_.size(); idx++)
{
+ const auto& trg = target_row_groups_[idx];
+ if (skip_read_range_excluded && trg.IsExcludedByReadRange()) {
+ continue;
+ }
+
+ if (trg.IsPartiallyMatched()) {
+ // Page-filtered RGs: only matching page byte ranges.
+ auto row_group_page_index_reader =
+ GetRowGroupPageIndexReader(trg.GetRowGroupIndex());
+ auto page_ranges =
PageFilteredRowGroupReader::ComputePageRanges(
+ trg, column_indices, row_group_page_index_reader,
+ file_reader_->parquet_reader());
+ ranges.insert(ranges.end(),
std::make_move_iterator(page_ranges.begin()),
+ std::make_move_iterator(page_ranges.end()));
+ } else {
+ // Fully-matched RGs: entire column chunk ranges.
+ auto rg_metadata =
file_metadata->RowGroup(trg.GetRowGroupIndex());
+ for (int32_t col_idx : column_indices) {
+ auto col_chunk = rg_metadata->ColumnChunk(col_idx);
+ int64_t offset = col_chunk->data_page_offset();
+ if (col_chunk->has_dictionary_page() &&
+ col_chunk->dictionary_page_offset() > 0 &&
+ offset > col_chunk->dictionary_page_offset()) {
+ offset = col_chunk->dictionary_page_offset();
+ }
+ ranges.push_back({offset,
col_chunk->total_compressed_size()});
}
- ranges.push_back({offset, col_chunk->total_compressed_size()});
}
}
+ return ranges;
}
- return ranges;
+
PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("FileReaderWrapper::DoCollectPreBufferRanges")
+}
+
+Result<std::vector<std::pair<uint64_t, uint64_t>>>
FileReaderWrapper::GetPreBufferRanges() {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<::arrow::io::ReadRange> ranges,
+ DoCollectPreBufferRanges(target_column_indices_,
+
/*skip_read_range_excluded=*/false,
+ /*start_idx=*/0));
+ std::vector<std::pair<uint64_t, uint64_t>> pre_buffer_ranges;
+ pre_buffer_ranges.reserve(ranges.size());
+ for (const auto& range : ranges) {
+ pre_buffer_ranges.emplace_back(static_cast<uint64_t>(range.offset),
+ static_cast<uint64_t>(range.length));
Review Comment:
[P1] Validate signed Parquet ranges before conversion
These Arrow ranges contain signed metadata values, but negative offsets or
`total_compressed_size` values are cast directly to very large `uint64_t`s.
`ReadAheadCache::Init()` then calls `CoalesceByteRanges()` before its bounds
checks; the combiner performs unchecked `offset + length` arithmetic and can
enter a pathological splitting loop, exhausting memory on a corrupt footer.
Please validate that offset and length are non-negative and that their sum does
not overflow before converting them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
