SteNicholas commented on code in PR #209:
URL: https://github.com/apache/paimon-cpp/pull/209#discussion_r3814147021
##########
src/paimon/common/utils/read_ahead_cache.cpp:
##########
@@ -89,38 +147,52 @@ 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()) {
+ // Entries are never evicted: the cache holds every published
Review Comment:
Please restore a hard cache-size bound. pre_buffer_limit only bounds each
prefetch window; subsequent reads prefetch later windows while every earlier
entry remains retained until the file closes. A sequential scan can therefore
cache all selected column chunks of a multi-GB Parquet file, especially since
buffer_size_limit was removed. Track in-flight operations separately and evict
completed ranges under a configurable total-byte limit.
--
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]
