SteNicholas commented on code in PR #209:
URL: https://github.com/apache/paimon-cpp/pull/209#discussion_r3809798641
##########
src/paimon/format/parquet/parquet_file_batch_reader.cpp:
##########
@@ -635,14 +635,18 @@ Result<std::vector<std::pair<uint64_t, uint64_t>>>
ParquetFileBatchReader::GenRe
PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("ParquetFileBatchReader::GenReadRanges")
}
+Result<std::vector<std::pair<uint64_t, uint64_t>>>
ParquetFileBatchReader::PreBufferRange() {
+ return reader_->GetPreBufferRanges();
+}
+
Result<::parquet::ReaderProperties>
ParquetFileBatchReader::CreateReaderProperties(
const std::shared_ptr<arrow::MemoryPool>& pool,
const std::map<std::string, std::string>& options) {
::parquet::ReaderProperties reader_properties;
// TODO(jinli.zjw): set more ReaderProperties (compare with java)
PAIMON_ASSIGN_OR_RAISE(
bool enable_pre_buffer,
- OptionsUtils::GetValueFromMap<bool>(options,
PARQUET_READ_ENABLE_PRE_BUFFER, true));
+ OptionsUtils::GetValueFromMap<bool>(options,
PARQUET_READ_ENABLE_PRE_BUFFER, false));
Review Comment:
[P1] Preserve prebuffering when the shared cache is inactive
This globally changes Parquet's prebuffer default to `false`, but
`ReadContext` defaults `EnablePrefetch` to `false`, so ordinary scans never
construct the shared cache and now have no prefetch layer. The same regression
occurs with `PrefetchCacheMode::NEVER` and excluded predicate/bitmap modes.
Please keep Arrow prebuffering enabled unless the shared cache will actually be
initialized.
##########
include/paimon/fs/file_system.h:
##########
@@ -155,7 +155,8 @@ class PAIMON_EXPORT BasicFileStatus {
}
/// Get the path of this file or directory.
- std::string GetPath() const {
+ /// @note The returned reference is valid as long as this object is alive.
+ const std::string& GetPath() const {
Review Comment:
[P2] Avoid changing the public `GetPath` return contract
Changing this public getter from `std::string` to `const std::string&` is
unrelated to the cache work and changes source/ABI behavior. It can also turn
previously safe code such as `const std::string& p = MakeStatus().GetPath()`
into a dangling reference because the owning status temporary is destroyed.
Please retain the value-returning API; the same applies to
`FileStatus::GetPath` below.
##########
src/paimon/common/utils/read_ahead_cache.cpp:
##########
@@ -193,31 +266,96 @@ void ReadAheadCache::Impl::Reset() {
is_cached_.clear();
pending_ranges_.clear();
is_initialized_ = false;
+ // The read/io counters are deliberately kept: a reader closed at EOF must
+ // still be able to report them through CollectMetrics().
+}
+
+void ReadAheadCache::Impl::CollectMetrics(std::shared_ptr<Metrics>* metrics)
const {
+ if (metrics == nullptr || !*metrics) {
+ return;
+ }
+ auto& m = *metrics;
+ m->SetCounter(ReadAheadCacheMetrics::READ_COUNT,
read_count_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_BYTES,
read_bytes_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_HITS,
hits_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES,
+ hit_bytes_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_MISSES,
misses_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_MISS_BYTES,
+ miss_bytes_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::IO_COUNT,
io_count_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::IO_BYTES,
io_bytes_.load(std::memory_order_relaxed));
+}
+
+void ReadAheadCache::Impl::Warmup() {
+ // Init() only registers the pending ranges; without this the first fetch
+ // starts when the first Read() arrives, racing the reader's own miss
fetch.
+ if (!pending_ranges_.empty()) {
+ PreBuffer(pending_ranges_.front().offset);
+ }
}
-Result<ByteSlice> ReadAheadCache::Impl::Read(const ByteRange& range) {
+std::vector<RangeCacheEntry> ReadAheadCache::Impl::FindCoveringEntries(const
ByteRange& range) {
+ std::vector<RangeCacheEntry> covering;
+ std::shared_lock<std::shared_mutex> lock(rw_mutex_);
+ // Find the entry holding the start of the range: the first entry whose
+ // end is beyond range.offset (entries are disjoint and sorted by offset).
+ auto it = std::lower_bound(entries_.begin(), entries_.end(), range.offset,
+ [](const RangeCacheEntry& e, uint64_t offset) {
+ return e.range.offset + e.range.length <=
offset;
+ });
+ if (it == entries_.end() || it->range.offset > range.offset) {
+ return covering;
+ }
+ if (it->range.Contains(range)) {
+ covering.push_back(*it);
+ return covering;
+ }
+ // The request spans several adjacent entries (a column chunk larger than
+ // one coalesced range): collect the contiguous run and check it covers
+ // the whole request. Entries exist from the moment their fetch is
+ // SUBMITTED, so a reader racing the prefetch waits for the in-flight
+ // fetch instead of issuing a second one for the same bytes.
+ uint64_t covered_end = it->range.offset + it->range.length;
+ covering.push_back(*it);
+ auto next = std::next(it);
+ while (covered_end < range.offset + range.length && next != entries_.end()
&&
+ next->range.offset == covered_end) {
+ covered_end = next->range.offset + next->range.length;
+ covering.push_back(*next);
+ ++next;
+ }
+ if (covered_end < range.offset + range.length) {
+ covering.clear();
+ }
+ return covering;
+}
+
+Result<bool> ReadAheadCache::Impl::Read(const ByteRange& range, char* dest) {
if (range.length == 0) {
- return ByteSlice{std::make_shared<Bytes>(0, memory_pool_.get()), 0, 0};
+ return true;
}
+ read_count_.fetch_add(1, std::memory_order_relaxed);
+ read_bytes_.fetch_add(range.length, std::memory_order_relaxed);
PreBuffer(range.offset);
Review Comment:
[P1] Publish in-flight entries before readers can miss
`PreBuffer` marks each range cached before `Cache` is called, but `Cache`
does not publish its entries until `MakeCacheEntries` has dispatched or
completed every `ReadAsync`. A concurrent reader can therefore see
`is_cached=true` while `FindCoveringEntries` still finds nothing, then issue a
duplicate underlying read. Please publish the promise-backed entries under the
lock before starting I/O, or otherwise expose the pending state to readers.
--
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]
