rok commented on code in PR #49855:
URL: https://github.com/apache/arrow/pull/49855#discussion_r3529687297
##########
cpp/src/parquet/file_reader.cc:
##########
Review Comment:
We could fall back to new read in case of miss?
```suggestion
// EvictPreBufferedDataBefore may have released the cached bytes
// which does not update the bitmap. A cache miss falls back to a
// direct read.
auto buffer = cached_source_->Read(col_range);
if (!buffer.ok() && !buffer.status().IsInvalid()) {
PARQUET_THROW_NOT_OK(buffer.status());
}
if (buffer.ok()) {
stream =
std::make_shared<::arrow::io::BufferReader>(*std::move(buffer));
}
}
if (stream == nullptr) {
```
##########
cpp/src/arrow/io/caching.cc:
##########
@@ -167,115 +175,154 @@ struct ReadRangeCache::Impl {
std::vector<RangeCacheEntry> new_entries;
new_entries.reserve(ranges.size());
for (const auto& range : ranges) {
- new_entries.emplace_back(range, file->ReadAsync(ctx, range.offset,
range.length,
-
/*allow_short_read=*/false));
+ new_entries.emplace_back(range, file->ReadAsync(ctx, range.offset,
range.length));
}
return new_entries;
}
- // Add the given ranges to the cache, coalescing them where possible
- virtual Status Cache(std::vector<ReadRange> ranges) {
+ // -- Public entry points (acquire entry_mutex, then delegate). --
+
+ // Add the given ranges to the cache, coalescing them where possible.
+ Status Cache(std::vector<ReadRange> ranges) {
ARROW_ASSIGN_OR_RAISE(
ranges, internal::CoalesceReadRanges(std::move(ranges),
options.hole_size_limit,
options.range_size_limit));
- std::vector<RangeCacheEntry> new_entries = MakeCacheEntries(ranges);
- // Add new entries, themselves ordered by offset
- if (entries.size() > 0) {
- std::vector<RangeCacheEntry> merged(entries.size() + new_entries.size());
- std::merge(entries.begin(), entries.end(), new_entries.begin(),
new_entries.end(),
- merged.begin());
- entries = std::move(merged);
- } else {
- entries = std::move(new_entries);
+ Status st;
+ {
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ std::vector<RangeCacheEntry> new_entries = MakeCacheEntries(ranges);
+ // Add new entries, themselves ordered by offset
+ if (entries.size() > 0) {
+ std::vector<RangeCacheEntry> merged(entries.size() +
new_entries.size());
+ std::merge(entries.begin(), entries.end(), new_entries.begin(),
new_entries.end(),
+ merged.begin());
+ entries = std::move(merged);
+ } else {
+ entries = std::move(new_entries);
+ }
}
- // Prefetch immediately, regardless of executor availability, if possible
- auto st = file->WillNeed(ranges);
+ // Prefetch immediately, regardless of executor availability, if possible.
+ // Do this outside the lock: WillNeed() may block on an mmap advise / I/O
+ // hint and we don't want to serialize concurrent Reads on it.
+ st = file->WillNeed(ranges);
// As this is optimisation only, I/O failures should not be treated as
fatal
if (st.IsIOError()) {
return Status::OK();
}
return st;
}
- // Read the given range from the cache, blocking if needed. Cannot read a
range
- // that spans cache entries.
- virtual Result<std::shared_ptr<Buffer>> Read(ReadRange range) {
+ // Read the given range from the cache, blocking if needed. Cannot read a
+ // range that spans cache entries.
+ Result<std::shared_ptr<Buffer>> Read(ReadRange range) {
if (range.length == 0) {
static const uint8_t byte = 0;
return std::make_shared<Buffer>(&byte, 0);
}
- const auto it = std::lower_bound(
- entries.begin(), entries.end(), range,
- [](const RangeCacheEntry& entry, const ReadRange& range) {
- return entry.range.offset + entry.range.length < range.offset +
range.length;
- });
- if (it != entries.end() && it->range.Contains(range)) {
- auto fut = MaybeRead(&*it);
- ARROW_ASSIGN_OR_RAISE(auto buf, fut.result());
+ Future<std::shared_ptr<Buffer>> fut;
+ int64_t slice_offset = 0;
+ {
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ const auto it = std::lower_bound(
+ entries.begin(), entries.end(), range,
+ [](const RangeCacheEntry& entry, const ReadRange& range) {
+ return entry.range.offset + entry.range.length < range.offset +
range.length;
+ });
+ if (it == entries.end() || !it->range.Contains(range)) {
+ return Status::Invalid("ReadRangeCache did not find matching cache
entry");
+ }
+ fut = MaybeRead(&*it);
+ slice_offset = range.offset - it->range.offset;
if (options.lazy && options.prefetch_limit > 0) {
int64_t num_prefetched = 0;
for (auto next_it = it + 1;
next_it != entries.end() && num_prefetched <
options.prefetch_limit;
++next_it) {
if (!next_it->future.is_valid()) {
next_it->future =
- file->ReadAsync(ctx, next_it->range.offset,
next_it->range.length,
- /*allow_short_read=*/false);
+ file->ReadAsync(ctx, next_it->range.offset,
next_it->range.length);
}
++num_prefetched;
}
}
- return SliceBuffer(std::move(buf), range.offset - it->range.offset,
range.length);
}
- return Status::Invalid("ReadRangeCache did not find matching cache entry");
+ // Drop the lock before blocking on the I/O future so other threads can
+ // still do lookups while a previously queued read is in flight.
+ ARROW_ASSIGN_OR_RAISE(auto buf, fut.result());
+ return SliceBuffer(std::move(buf), slice_offset, range.length);
}
- virtual Future<> Wait() {
+ Future<> Wait() {
std::vector<Future<>> futures;
- for (auto& entry : entries) {
- futures.emplace_back(MaybeRead(&entry));
+ {
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ futures.reserve(entries.size());
+ for (auto& entry : entries) {
+ futures.emplace_back(MaybeRead(&entry));
+ }
}
return AllComplete(futures);
}
+ // `entries` is sorted by offset; an entry that extends past `end_offset`
+ // (coalesced with a range a later consumer still needs) is kept.
+ int64_t EvictEntriesBefore(int64_t end_offset) {
+ int64_t n_evicted = 0;
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ auto it = entries.begin();
+ while (it != entries.end() && it->range.offset < end_offset) {
+ if (it->range.offset + it->range.length <= end_offset) {
+ it = entries.erase(it);
+ ++n_evicted;
+ } else {
+ ++it;
+ }
+ }
Review Comment:
```suggestion
// remove_if is stable, so the ordering of the kept entries (sorted by
// offset) is preserved.
int64_t EvictEntriesBefore(int64_t end_offset) {
std::unique_lock<std::mutex> guard(entry_mutex);
const auto first_kept = std::remove_if(
entries.begin(), entries.end(), [end_offset](const RangeCacheEntry&
entry) {
return entry.range.offset + entry.range.length <= end_offset;
});
const auto n_evicted = static_cast<int64_t>(entries.end() - first_kept);
entries.erase(first_kept, entries.end());
```
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1275,10 +1239,31 @@
FileReaderImpl::GetRecordBatchGenerator(std::shared_ptr<FileReader> reader,
reader_properties_.cache_options());
END_PARQUET_CATCH_EXCEPTIONS
}
+ // GH-39808: evict each row group's bytes as the decoded prefix advances, so
+ // memory stays bounded. Only this read-once path evicts, so PreBuffer()'s
+ // contract is unchanged for other callers.
+ std::shared_ptr<ReadCacheEvictionState> eviction_state;
+ if (reader_properties_.pre_buffer() && !column_indices.empty() &&
+ !row_group_indices.empty()) {
+ const int64_t kNoMoreRanges = std::numeric_limits<int64_t>::max();
+ std::vector<int64_t> evict_before(row_group_indices.size() + 1,
kNoMoreRanges);
+ for (int64_t i = static_cast<int64_t>(row_group_indices.size()) - 1; i >=
0; --i) {
Review Comment:
What happens if row_group_indices isn't ascending in file order (e.g.
{3,2,1,0})?
--
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]
