github-actions[bot] commented on code in PR #66548:
URL: https://github.com/apache/doris/pull/66548#discussion_r3744234885
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -107,6 +113,428 @@ bvar::Adder<uint64_t>
g_peer_cross_compute_group_read("peer_cross_compute_group_
bvar::Adder<uint64_t>
g_peer_same_compute_group_read("peer_same_compute_group_read");
bvar::Adder<uint64_t> g_peer_lazy_fetch_triggered("peer_lazy_fetch_triggered");
+FileScannerV2ReaderLocalCache::FileScannerV2ReaderLocalCache(
+ size_t capacity, std::shared_ptr<doris::MemTrackerLimiter>
query_mem_tracker)
+ : _capacity(capacity),
+ _query_mem_tracker(std::move(query_mem_tracker)),
+
_memory_tracker(std::make_shared<doris::MemTracker>("FileScannerV2ReaderLocalCache"))
{}
+
+FileScannerV2ReaderLocalCache::~FileScannerV2ReaderLocalCache() {
Review Comment:
[P1] Keep the cache destructor allocation-free
This destructor is implicitly `noexcept`, but `_file_caches()` allocates a
new vector and calls `reserve(_files.size())` before any of the exception-safe
drain logic runs. Under the same query/global memory pressure this cache is
designed to tolerate, `std::bad_alloc` here escapes the destructor and
terminates the BE. The registry also retains expired weak entries, so the
snapshot can allocate according to the scanner's file-count high-water mark
even when no live file cache remains. Please make teardown walk/drain the
registry without allocating, or contain allocation failure while still
releasing the budget, and add a teardown-failure regression test.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1071,17 +1748,27 @@ Status
CachedRemoteFileReader::_read_from_indirect_cache(size_t offset, Slice re
stats.cache_get_or_set_timer += sw.elapsed_time();
auto empty_blocks = _collect_remote_read_blocks(holder, stats);
- size_t empty_start = 0;
- size_t empty_end = 0;
- PeerFetchedBlockSet peer_fetched_blocks;
- RETURN_IF_ERROR(_read_remote_blocks_into_cache(empty_blocks, offset,
bytes_req, already_read,
- result, is_dryrun, stats,
source_read_breakdown,
- io_ctx,
indirect_read_bytes, empty_start,
- empty_end,
peer_fetched_blocks));
+ PeerFetchedBlockSet fetched_blocks;
+ size_t run_start = 0;
+ for (size_t index = 1; index <= empty_blocks.size(); ++index) {
+ const bool end_of_run =
+ index == empty_blocks.size() ||
+ empty_blocks[index - 1]->range().right + 1 !=
empty_blocks[index]->range().left;
+ if (!end_of_run) {
+ continue;
+ }
+ // A cache hit is a hard merge boundary. Reading across it would
redownload resident data
+ // and violate the cache-aware miss coalescing invariant used by
StarRocks.
+ std::vector<FileBlockSPtr> contiguous_misses(empty_blocks.begin() +
run_start,
Review Comment:
[P2] Account for every split source request
This loop can now issue one peer/S3 call per contiguous miss run (the new
miss-hit-miss test observes two), but the shared stats retain only copied byte
totals and one last-writer-wins `from_peer_cache` bit. `_update_stats()`
therefore increments `NumRemoteIOTotal`/`NumPeerIOTotal` at most once, and a
later run can erase the source of a backward-aligned fill that copied no caller
bytes. That hides the request amplification this change is meant to control and
can omit mixed peer/S3 work. Please accumulate per-run peer/remote call counts
(and used-source state) and assert both counts in split and mixed-source tests.
##########
be/src/io/fs/buffered_reader.cpp:
##########
@@ -397,6 +494,39 @@ Status MergeRangeFileReader::_fill_box(int range_index,
size_t start_offset, siz
return Status::OK();
}
+void MergeRangeFileReader::_record_merged_read(int range_index, size_t
start_offset,
+ size_t bytes_read) {
+ if (bytes_read == 0) {
+ return;
+ }
+ if (range_index < 0) {
+ _statistics.merged_useful_bytes += bytes_read;
+ return;
+ }
+ const size_t read_end = start_offset + bytes_read;
+ size_t useful_bytes = 0;
+ size_t future_predicate_bytes = 0;
+ for (size_t index = static_cast<size_t>(range_index);
+ index < _random_access_ranges.size() &&
+ _random_access_ranges[index].start_offset < read_end;
+ ++index) {
+ const auto& range = _random_access_ranges[index];
+ const size_t overlap_start = std::max(start_offset,
range.start_offset);
+ const size_t overlap_end = std::min(read_end, range.end_offset);
+ if (overlap_start >= overlap_end) {
+ continue;
+ }
+ const size_t overlap = overlap_end - overlap_start;
+ useful_bytes += overlap;
Review Comment:
[P2] Count the union of eager-range overlaps
Eager native ranges are only sorted before this reader is constructed;
unlike deferred ranges, they are not coalesced. For pre-PARQUET-816 files, the
100-byte chunk padding can make adjacent ranges overlap, so this loop counts
those bytes twice and `bytes_read - useful_bytes` underflows. `MergedGapBytes`
can then become negative or enormous, breaking the new profile invariant.
Please normalize constructor ranges too, or compute the union of intersections
and guard the subtraction; add an eager-overlap accounting test.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -107,6 +113,428 @@ bvar::Adder<uint64_t>
g_peer_cross_compute_group_read("peer_cross_compute_group_
bvar::Adder<uint64_t>
g_peer_same_compute_group_read("peer_same_compute_group_read");
bvar::Adder<uint64_t> g_peer_lazy_fetch_triggered("peer_lazy_fetch_triggered");
+FileScannerV2ReaderLocalCache::FileScannerV2ReaderLocalCache(
+ size_t capacity, std::shared_ptr<doris::MemTrackerLimiter>
query_mem_tracker)
+ : _capacity(capacity),
+ _query_mem_tracker(std::move(query_mem_tracker)),
+
_memory_tracker(std::make_shared<doris::MemTracker>("FileScannerV2ReaderLocalCache"))
{}
+
+FileScannerV2ReaderLocalCache::~FileScannerV2ReaderLocalCache() {
+ auto files = _file_caches();
+ for (const auto& file : files) {
+ file->_drain(this);
+ }
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_memory_bytes == 0);
+ DORIS_CHECK(_reserved_bytes == 0);
+}
+
+std::shared_ptr<FileScannerV2ReaderLocalFileCache>
+FileScannerV2ReaderLocalCache::create_file_cache() {
+ if (_capacity == 0) {
+ return nullptr;
+ }
+ std::shared_ptr<FileScannerV2ReaderLocalFileCache> file_cache;
+ try {
+ file_cache = std::shared_ptr<FileScannerV2ReaderLocalFileCache>(
+ new FileScannerV2ReaderLocalFileCache(shared_from_this()));
+ } catch (const doris::Exception&) {
+ return nullptr;
+ } catch (const std::bad_alloc&) {
+ return nullptr;
+ }
+ {
+ std::lock_guard lock(_registry_mutex);
+ try {
+ std::erase_if(_files, [](const auto& file) { return
file.expired(); });
+ _files.emplace_back(file_cache);
+ } catch (const doris::Exception&) {
+ return nullptr;
+ } catch (const std::bad_alloc&) {
+ return nullptr;
+ }
+ }
+ return file_cache;
+}
+
+bool FileScannerV2ReaderLocalCache::_try_reserve(size_t bytes) {
+ std::lock_guard lock(_budget_mutex);
+ if (bytes > _capacity) {
+ return false;
+ }
+ if (_memory_bytes + _reserved_bytes + bytes > _capacity) {
+ return false;
+ }
+ if (_query_mem_tracker != nullptr && _query_mem_tracker->limit() >= 0 &&
+ _query_mem_tracker->consumption() + cast_set<int64_t>(_reserved_bytes
+ bytes) >
+ _query_mem_tracker->limit()) {
+ return false;
+ }
+ if (GlobalMemoryArbitrator::is_exceed_soft_mem_limit(
+ cast_set<int64_t>(_reserved_bytes + bytes))) {
+ return false;
+ }
+ _reserved_bytes += bytes;
+ return true;
+}
+
+bool FileScannerV2ReaderLocalCache::_reserve(size_t bytes,
+
FileScannerV2ReaderLocalFileCache* requester,
+ size_t* evicted) {
+ if (_try_reserve(bytes)) {
+ return true;
+ }
+ // A stream may recycle its own cold blocks, but it must never evict
another stream's hot
+ // block map. StarRocks gets the same isolation from
CacheInputStream::_block_map ownership.
+ while (requester->_evict_one()) {
+ ++*evicted;
+ if (_try_reserve(bytes)) {
+ return true;
+ }
+ }
+ return false;
+}
+
+void FileScannerV2ReaderLocalCache::_commit(size_t bytes) {
+ {
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_reserved_bytes >= bytes);
+ _reserved_bytes -= bytes;
+ _memory_bytes += bytes;
+ }
+ _memory_tracker->consume(cast_set<int64_t>(bytes));
+}
+
+void FileScannerV2ReaderLocalCache::_cancel_reservation(size_t bytes) {
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_reserved_bytes >= bytes);
+ _reserved_bytes -= bytes;
+}
+
+void FileScannerV2ReaderLocalCache::_release(size_t bytes) {
+ {
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_memory_bytes >= bytes);
+ _memory_bytes -= bytes;
+ }
+ _memory_tracker->release(cast_set<int64_t>(bytes));
+}
+
+std::vector<std::shared_ptr<FileScannerV2ReaderLocalFileCache>>
+FileScannerV2ReaderLocalCache::_file_caches() const {
+ std::vector<std::shared_ptr<FileScannerV2ReaderLocalFileCache>> files;
+ std::lock_guard lock(_registry_mutex);
+ files.reserve(_files.size());
+ for (const auto& file : _files) {
+ if (auto live_file = file.lock(); live_file != nullptr) {
+ files.push_back(std::move(live_file));
+ }
+ }
+ return files;
+}
+
+size_t FileScannerV2ReaderLocalCache::entry_count() const {
+ size_t count = 0;
+ for (const auto& file : _file_caches()) {
+ count += file->entry_count();
+ }
+ return count;
+}
+
+size_t FileScannerV2ReaderLocalCache::memory_usage() const {
+ std::lock_guard lock(_budget_mutex);
+ return _memory_bytes;
+}
+
+int64_t FileScannerV2ReaderLocalCache::tracked_memory() const {
+ return _memory_tracker->consumption();
+}
+
+FileScannerV2ReaderLocalFileCache::FileScannerV2ReaderLocalFileCache(
+ std::shared_ptr<FileScannerV2ReaderLocalCache> owner)
+ : _owner(std::move(owner)) {}
+
+FileScannerV2ReaderLocalFileCache::~FileScannerV2ReaderLocalFileCache() {
+ if (auto owner = _owner.lock(); owner != nullptr) {
+ _drain(owner.get());
+ }
+}
+
+void FileScannerV2ReaderLocalFileCache::_drain(FileScannerV2ReaderLocalCache*
owner) {
+ DORIS_CHECK(owner != nullptr);
+ size_t memory_bytes = 0;
+ size_t reserved_bytes = 0;
+ auto clear_entries = [&]() {
+ std::unique_lock lock(_mutex);
+ for (auto& [_, entry] : _entries) {
+ if (entry->data != nullptr) {
+ memory_bytes += entry->data->size();
+ }
+ reserved_bytes += entry->reserved_bytes;
+ }
+ _entries.clear();
+ _lru.clear();
+ };
+ try {
+ std::optional<SwitchThreadMemTrackerLimiter> switch_query_tracker;
+ if (owner->_query_mem_tracker != nullptr) {
+ switch_query_tracker.emplace(owner->_query_mem_tracker);
+ }
+ clear_entries();
+ } catch (...) {
+ // Destructors cannot propagate memory-tracker setup failures during
query cancellation.
+ clear_entries();
+ }
+ if (memory_bytes > 0) {
+ owner->_release(memory_bytes);
+ }
+ if (reserved_bytes > 0) {
+ owner->_cancel_reservation(reserved_bytes);
+ }
+}
+
+void FileScannerV2ReaderLocalFileCache::_touch_locked(const
std::shared_ptr<Entry>& entry) {
+ if (entry->in_lru) {
+ _lru.splice(_lru.begin(), _lru, entry->lru_position);
+ }
+}
+
+bool FileScannerV2ReaderLocalFileCache::_evict_one() {
+ const auto owner = _owner.lock();
+ if (owner == nullptr) {
+ return false;
+ }
+ std::shared_ptr<std::vector<char>> data;
+ {
+ std::lock_guard lock(_mutex);
+ size_t candidates = _lru.size();
+ while (candidates-- > 0 && !_lru.empty()) {
+ const size_t victim_offset = _lru.back();
+ const auto victim = _entries.find(victim_offset);
+ DORIS_CHECK(victim != _entries.end());
+ DORIS_CHECK(!victim->second->loading);
+ if (victim->second->data.use_count() > 1) {
+ // Keep pinned blocks discoverable so another reader cannot
start a duplicate fill.
+ _lru.splice(_lru.begin(), _lru, victim->second->lru_position);
+ continue;
+ }
+ data = std::move(victim->second->data);
+ _lru.pop_back();
+ _entries.erase(victim);
+ break;
+ }
+ }
+ if (data == nullptr) {
+ return false;
+ }
+ const size_t bytes = data->size();
+ // Eviction and destruction are noexcept cleanup paths. A stack guard
avoids a second heap
+ // allocation while releasing memory under pressure; if tracker switching
itself fails, the
+ // block is still released and the explicit cache budget remains
consistent.
+ try {
+ std::optional<SwitchThreadMemTrackerLimiter> switch_query_tracker;
+ if (owner->_query_mem_tracker != nullptr) {
+ switch_query_tracker.emplace(owner->_query_mem_tracker);
+ }
+ data.reset();
+ } catch (...) {
+ data.reset();
+ }
+ owner->_release(bytes);
+ return true;
+}
+
+void FileScannerV2ReaderLocalFileCache::_abort_load(size_t block_offset,
+ const
std::shared_ptr<Entry>& entry) {
+ size_t reserved_bytes = 0;
+ {
+ std::unique_lock lock(_mutex);
+ reserved_bytes = entry->reserved_bytes;
+ entry->reserved_bytes = 0;
+ entry->loading = false;
+ const auto it = _entries.find(block_offset);
+ if (it != _entries.end() && it->second == entry) {
+ _entries.erase(it);
+ }
+ }
+ if (reserved_bytes != 0) {
+ if (const auto owner = _owner.lock(); owner != nullptr) {
+ owner->_cancel_reservation(reserved_bytes);
+ }
+ }
+ // A loader must publish every exit, including allocation and tracker
exceptions, otherwise a
+ // same-block waiter can remain asleep after the scan has already fallen
back to FileCache.
+ entry->ready.notify_all();
+}
+
+bool FileScannerV2ReaderLocalFileCache::pin_if_present(size_t block_offset,
size_t read_offset,
+ size_t read_size,
LookupResult* lookup) {
+ DORIS_CHECK(lookup != nullptr);
+ *lookup = {};
+ std::shared_ptr<Entry> entry;
+ bool touch_lru = false;
+ {
+ std::shared_lock lock(_mutex);
+ const auto it = _entries.find(block_offset);
+ if (it == _entries.end() || it->second->loading ||
!it->second->load_status.ok() ||
+ it->second->data == nullptr || read_offset < block_offset ||
+ read_offset - block_offset > it->second->data->size() ||
+ read_size > it->second->data->size() - (read_offset -
block_offset)) {
+ return false;
+ }
+ entry = it->second;
+ lookup->data = entry->data;
+ lookup->admitted = true;
+ lookup->hit = true;
+ touch_lru =
+ entry->hit_count.fetch_add(1, std::memory_order_relaxed) %
LRU_TOUCH_INTERVAL == 0;
+ if (touch_lru) {
+ lookup->file_block_to_touch = entry->source_file_block.lock();
+ }
+ }
+ if (touch_lru) {
+ std::unique_lock lock(_mutex);
+ const auto it = _entries.find(block_offset);
+ if (it != _entries.end() && it->second == entry) {
+ _touch_locked(entry);
+ }
+ }
+ return true;
+}
+
+bool FileScannerV2ReaderLocalFileCache::read_if_present(size_t block_offset,
size_t read_offset,
+ Slice result,
LookupResult* lookup) {
+ if (!pin_if_present(block_offset, read_offset, result.size, lookup)) {
+ return false;
+ }
+ memcpy(result.data, lookup->data->data() + read_offset - block_offset,
result.size);
+ return true;
+}
+
+Status FileScannerV2ReaderLocalFileCache::get_or_load(size_t block_offset,
size_t block_size,
+ const FileBlockSPtr&
file_block,
+ size_t file_block_offset,
+ LookupResult* lookup) {
+ DORIS_CHECK(lookup != nullptr);
+ *lookup = {};
+ const auto owner = _owner.lock();
+ if (owner == nullptr) {
+ return Status::OK();
+ }
+ std::shared_ptr<Entry> entry;
+ bool load = false;
+ {
+ std::unique_lock lock(_mutex);
+ const auto it = _entries.find(block_offset);
+ if (it == _entries.end()) {
+ try {
+ entry = std::make_shared<Entry>();
+ } catch (const doris::Exception&) {
+ return Status::OK();
+ } catch (const std::bad_alloc&) {
+ return Status::OK();
+ }
+ try {
+ _entries.emplace(block_offset, entry);
+ } catch (const doris::Exception&) {
+ return Status::OK();
+ } catch (const std::bad_alloc&) {
+ return Status::OK();
+ }
+ load = true;
+ } else {
+ entry = it->second;
+ lookup->admitted = true;
+ if (entry->loading) {
+ lookup->waited = true;
+
TEST_SYNC_POINT("CachedRemoteFileReader::reader_local_cache_before_wait");
+ MonotonicStopWatch wait_watch;
+ wait_watch.start();
+ entry->ready.wait(lock, [&entry]() { return !entry->loading;
});
+ lookup->wait_time = wait_watch.elapsed_time();
+ }
+ RETURN_IF_ERROR(entry->load_status);
+ if (entry->data == nullptr || entry->data->size() < block_size) {
Review Comment:
[P2] Let a wider extent replace a short cache entry
This safely refuses to reuse a short promotion, but it leaves that entry in
`_entries` and `_lru`. If FileCache later recreates a wider cell with the same
left boundary, every wider read hits this same mismatch and falls back to disk
forever; the wider extent can never be promoted until unrelated capacity
eviction happens. Please retire/replace an unpinned undersized entry (or
include the extent in the key) while preserving single-flight behavior, and
cover short-cell-then-wider-cell repeated reads.
##########
be/benchmark/parquet/benchmark_parquet_decoder.hpp:
##########
@@ -626,7 +625,7 @@ inline void run_decoder(benchmark::State& state,
DecoderScenario scenario, int s
inline bool register_decoder_benchmarks() {
for (const auto& scenario : decoder_scenarios()) {
- for (const int selectivity : {0, 1, 10, 50, 90, 100}) {
+ for (const int selectivity : {0, 1, 5, 10, 50, 90, 100}) {
Review Comment:
[P2] Keep the benchmark validation guide in sync
This loop now registers 19 x 7 x 2 = 266 decoder cases, and the changed
invariant test expects 266, but `be/benchmark/parquet/AGENTS.md` still tells
reviewers to expect 228 in three places and still lists only six selectivities.
That guide is the mandatory validation contract for this directory, so its
prescribed registration check will reject the new matrix. Please update the
count and add the 5% axis there as part of this change.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -107,6 +113,428 @@ bvar::Adder<uint64_t>
g_peer_cross_compute_group_read("peer_cross_compute_group_
bvar::Adder<uint64_t>
g_peer_same_compute_group_read("peer_same_compute_group_read");
bvar::Adder<uint64_t> g_peer_lazy_fetch_triggered("peer_lazy_fetch_triggered");
+FileScannerV2ReaderLocalCache::FileScannerV2ReaderLocalCache(
+ size_t capacity, std::shared_ptr<doris::MemTrackerLimiter>
query_mem_tracker)
+ : _capacity(capacity),
+ _query_mem_tracker(std::move(query_mem_tracker)),
+
_memory_tracker(std::make_shared<doris::MemTracker>("FileScannerV2ReaderLocalCache"))
{}
+
+FileScannerV2ReaderLocalCache::~FileScannerV2ReaderLocalCache() {
+ auto files = _file_caches();
+ for (const auto& file : files) {
+ file->_drain(this);
+ }
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_memory_bytes == 0);
+ DORIS_CHECK(_reserved_bytes == 0);
+}
+
+std::shared_ptr<FileScannerV2ReaderLocalFileCache>
+FileScannerV2ReaderLocalCache::create_file_cache() {
+ if (_capacity == 0) {
+ return nullptr;
+ }
+ std::shared_ptr<FileScannerV2ReaderLocalFileCache> file_cache;
+ try {
+ file_cache = std::shared_ptr<FileScannerV2ReaderLocalFileCache>(
+ new FileScannerV2ReaderLocalFileCache(shared_from_this()));
+ } catch (const doris::Exception&) {
+ return nullptr;
+ } catch (const std::bad_alloc&) {
+ return nullptr;
+ }
+ {
+ std::lock_guard lock(_registry_mutex);
+ try {
+ std::erase_if(_files, [](const auto& file) { return
file.expired(); });
+ _files.emplace_back(file_cache);
+ } catch (const doris::Exception&) {
+ return nullptr;
+ } catch (const std::bad_alloc&) {
+ return nullptr;
+ }
+ }
+ return file_cache;
+}
+
+bool FileScannerV2ReaderLocalCache::_try_reserve(size_t bytes) {
+ std::lock_guard lock(_budget_mutex);
+ if (bytes > _capacity) {
+ return false;
+ }
+ if (_memory_bytes + _reserved_bytes + bytes > _capacity) {
+ return false;
+ }
+ if (_query_mem_tracker != nullptr && _query_mem_tracker->limit() >= 0 &&
+ _query_mem_tracker->consumption() + cast_set<int64_t>(_reserved_bytes
+ bytes) >
+ _query_mem_tracker->limit()) {
+ return false;
+ }
+ if (GlobalMemoryArbitrator::is_exceed_soft_mem_limit(
+ cast_set<int64_t>(_reserved_bytes + bytes))) {
+ return false;
+ }
+ _reserved_bytes += bytes;
+ return true;
+}
+
+bool FileScannerV2ReaderLocalCache::_reserve(size_t bytes,
+
FileScannerV2ReaderLocalFileCache* requester,
+ size_t* evicted) {
+ if (_try_reserve(bytes)) {
+ return true;
+ }
+ // A stream may recycle its own cold blocks, but it must never evict
another stream's hot
+ // block map. StarRocks gets the same isolation from
CacheInputStream::_block_map ownership.
+ while (requester->_evict_one()) {
+ ++*evicted;
+ if (_try_reserve(bytes)) {
+ return true;
+ }
+ }
+ return false;
+}
+
+void FileScannerV2ReaderLocalCache::_commit(size_t bytes) {
+ {
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_reserved_bytes >= bytes);
+ _reserved_bytes -= bytes;
+ _memory_bytes += bytes;
+ }
+ _memory_tracker->consume(cast_set<int64_t>(bytes));
+}
+
+void FileScannerV2ReaderLocalCache::_cancel_reservation(size_t bytes) {
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_reserved_bytes >= bytes);
+ _reserved_bytes -= bytes;
+}
+
+void FileScannerV2ReaderLocalCache::_release(size_t bytes) {
+ {
+ std::lock_guard lock(_budget_mutex);
+ DORIS_CHECK(_memory_bytes >= bytes);
+ _memory_bytes -= bytes;
+ }
+ _memory_tracker->release(cast_set<int64_t>(bytes));
+}
+
+std::vector<std::shared_ptr<FileScannerV2ReaderLocalFileCache>>
+FileScannerV2ReaderLocalCache::_file_caches() const {
+ std::vector<std::shared_ptr<FileScannerV2ReaderLocalFileCache>> files;
+ std::lock_guard lock(_registry_mutex);
+ files.reserve(_files.size());
+ for (const auto& file : _files) {
+ if (auto live_file = file.lock(); live_file != nullptr) {
+ files.push_back(std::move(live_file));
+ }
+ }
+ return files;
+}
+
+size_t FileScannerV2ReaderLocalCache::entry_count() const {
+ size_t count = 0;
+ for (const auto& file : _file_caches()) {
+ count += file->entry_count();
+ }
+ return count;
+}
+
+size_t FileScannerV2ReaderLocalCache::memory_usage() const {
+ std::lock_guard lock(_budget_mutex);
+ return _memory_bytes;
+}
+
+int64_t FileScannerV2ReaderLocalCache::tracked_memory() const {
+ return _memory_tracker->consumption();
+}
+
+FileScannerV2ReaderLocalFileCache::FileScannerV2ReaderLocalFileCache(
+ std::shared_ptr<FileScannerV2ReaderLocalCache> owner)
+ : _owner(std::move(owner)) {}
+
+FileScannerV2ReaderLocalFileCache::~FileScannerV2ReaderLocalFileCache() {
+ if (auto owner = _owner.lock(); owner != nullptr) {
+ _drain(owner.get());
+ }
+}
+
+void FileScannerV2ReaderLocalFileCache::_drain(FileScannerV2ReaderLocalCache*
owner) {
+ DORIS_CHECK(owner != nullptr);
+ size_t memory_bytes = 0;
+ size_t reserved_bytes = 0;
+ auto clear_entries = [&]() {
+ std::unique_lock lock(_mutex);
+ for (auto& [_, entry] : _entries) {
+ if (entry->data != nullptr) {
+ memory_bytes += entry->data->size();
+ }
+ reserved_bytes += entry->reserved_bytes;
+ }
+ _entries.clear();
+ _lru.clear();
+ };
+ try {
+ std::optional<SwitchThreadMemTrackerLimiter> switch_query_tracker;
+ if (owner->_query_mem_tracker != nullptr) {
+ switch_query_tracker.emplace(owner->_query_mem_tracker);
+ }
+ clear_entries();
+ } catch (...) {
+ // Destructors cannot propagate memory-tracker setup failures during
query cancellation.
+ clear_entries();
+ }
+ if (memory_bytes > 0) {
+ owner->_release(memory_bytes);
+ }
+ if (reserved_bytes > 0) {
+ owner->_cancel_reservation(reserved_bytes);
+ }
+}
+
+void FileScannerV2ReaderLocalFileCache::_touch_locked(const
std::shared_ptr<Entry>& entry) {
+ if (entry->in_lru) {
+ _lru.splice(_lru.begin(), _lru, entry->lru_position);
+ }
+}
+
+bool FileScannerV2ReaderLocalFileCache::_evict_one() {
+ const auto owner = _owner.lock();
+ if (owner == nullptr) {
+ return false;
+ }
+ std::shared_ptr<std::vector<char>> data;
+ {
+ std::lock_guard lock(_mutex);
+ size_t candidates = _lru.size();
+ while (candidates-- > 0 && !_lru.empty()) {
+ const size_t victim_offset = _lru.back();
+ const auto victim = _entries.find(victim_offset);
+ DORIS_CHECK(victim != _entries.end());
+ DORIS_CHECK(!victim->second->loading);
+ if (victim->second->data.use_count() > 1) {
+ // Keep pinned blocks discoverable so another reader cannot
start a duplicate fill.
+ _lru.splice(_lru.begin(), _lru, victim->second->lru_position);
+ continue;
+ }
+ data = std::move(victim->second->data);
+ _lru.pop_back();
+ _entries.erase(victim);
+ break;
+ }
+ }
+ if (data == nullptr) {
+ return false;
+ }
+ const size_t bytes = data->size();
+ // Eviction and destruction are noexcept cleanup paths. A stack guard
avoids a second heap
+ // allocation while releasing memory under pressure; if tracker switching
itself fails, the
+ // block is still released and the explicit cache budget remains
consistent.
+ try {
+ std::optional<SwitchThreadMemTrackerLimiter> switch_query_tracker;
+ if (owner->_query_mem_tracker != nullptr) {
+ switch_query_tracker.emplace(owner->_query_mem_tracker);
+ }
+ data.reset();
+ } catch (...) {
+ data.reset();
+ }
+ owner->_release(bytes);
+ return true;
+}
+
+void FileScannerV2ReaderLocalFileCache::_abort_load(size_t block_offset,
+ const
std::shared_ptr<Entry>& entry) {
+ size_t reserved_bytes = 0;
+ {
+ std::unique_lock lock(_mutex);
+ reserved_bytes = entry->reserved_bytes;
+ entry->reserved_bytes = 0;
+ entry->loading = false;
+ const auto it = _entries.find(block_offset);
+ if (it != _entries.end() && it->second == entry) {
+ _entries.erase(it);
+ }
+ }
+ if (reserved_bytes != 0) {
+ if (const auto owner = _owner.lock(); owner != nullptr) {
+ owner->_cancel_reservation(reserved_bytes);
+ }
+ }
+ // A loader must publish every exit, including allocation and tracker
exceptions, otherwise a
+ // same-block waiter can remain asleep after the scan has already fallen
back to FileCache.
+ entry->ready.notify_all();
+}
+
+bool FileScannerV2ReaderLocalFileCache::pin_if_present(size_t block_offset,
size_t read_offset,
+ size_t read_size,
LookupResult* lookup) {
+ DORIS_CHECK(lookup != nullptr);
+ *lookup = {};
+ std::shared_ptr<Entry> entry;
+ bool touch_lru = false;
+ {
+ std::shared_lock lock(_mutex);
+ const auto it = _entries.find(block_offset);
+ if (it == _entries.end() || it->second->loading ||
!it->second->load_status.ok() ||
+ it->second->data == nullptr || read_offset < block_offset ||
+ read_offset - block_offset > it->second->data->size() ||
+ read_size > it->second->data->size() - (read_offset -
block_offset)) {
+ return false;
+ }
+ entry = it->second;
+ lookup->data = entry->data;
+ lookup->admitted = true;
+ lookup->hit = true;
+ touch_lru =
+ entry->hit_count.fetch_add(1, std::memory_order_relaxed) %
LRU_TOUCH_INTERVAL == 0;
+ if (touch_lru) {
+ lookup->file_block_to_touch = entry->source_file_block.lock();
+ }
+ }
+ if (touch_lru) {
+ std::unique_lock lock(_mutex);
+ const auto it = _entries.find(block_offset);
+ if (it != _entries.end() && it->second == entry) {
+ _touch_locked(entry);
+ }
+ }
+ return true;
+}
+
+bool FileScannerV2ReaderLocalFileCache::read_if_present(size_t block_offset,
size_t read_offset,
+ Slice result,
LookupResult* lookup) {
+ if (!pin_if_present(block_offset, read_offset, result.size, lookup)) {
+ return false;
+ }
+ memcpy(result.data, lookup->data->data() + read_offset - block_offset,
result.size);
+ return true;
+}
+
+Status FileScannerV2ReaderLocalFileCache::get_or_load(size_t block_offset,
size_t block_size,
+ const FileBlockSPtr&
file_block,
+ size_t file_block_offset,
+ LookupResult* lookup) {
+ DORIS_CHECK(lookup != nullptr);
+ *lookup = {};
+ const auto owner = _owner.lock();
+ if (owner == nullptr) {
+ return Status::OK();
+ }
+ std::shared_ptr<Entry> entry;
+ bool load = false;
+ {
+ std::unique_lock lock(_mutex);
+ const auto it = _entries.find(block_offset);
+ if (it == _entries.end()) {
+ try {
+ entry = std::make_shared<Entry>();
+ } catch (const doris::Exception&) {
+ return Status::OK();
+ } catch (const std::bad_alloc&) {
+ return Status::OK();
+ }
+ try {
+ _entries.emplace(block_offset, entry);
+ } catch (const doris::Exception&) {
+ return Status::OK();
+ } catch (const std::bad_alloc&) {
+ return Status::OK();
+ }
+ load = true;
+ } else {
+ entry = it->second;
+ lookup->admitted = true;
+ if (entry->loading) {
+ lookup->waited = true;
+
TEST_SYNC_POINT("CachedRemoteFileReader::reader_local_cache_before_wait");
+ MonotonicStopWatch wait_watch;
+ wait_watch.start();
+ entry->ready.wait(lock, [&entry]() { return !entry->loading;
});
+ lookup->wait_time = wait_watch.elapsed_time();
+ }
+ RETURN_IF_ERROR(entry->load_status);
+ if (entry->data == nullptr || entry->data->size() < block_size) {
+ // FileCache ranges can end at different boundaries for the
same aligned offset.
+ // Never reuse a shorter promotion for a later, wider range.
+ lookup->admitted = false;
+ return Status::OK();
+ }
+ lookup->hit = true;
+ lookup->data = entry->data;
+ _touch_locked(entry);
+ }
+ }
+
+ if (!load) {
+ return Status::OK();
+ }
+
+ std::shared_ptr<std::vector<char>> data;
+ std::optional<SwitchThreadMemTrackerLimiter> switch_query_tracker;
+ try {
+ if (!owner->_reserve(block_size, this, &lookup->evicted)) {
Review Comment:
[P2] Publish reservation failures as admission rejects
`LookupResult::admission_rejected` defaults to false and is never set
anywhere, including this `_reserve()` failure path. `_read_local_block()`
consequently adds zero to `ReaderLocalCacheAdmissionRejects` for every
capacity, query-limit, and global-soft-limit rejection, making the new counter
permanently dead. Please set the flag for policy rejection (and define whether
allocation failure is included), then assert it in the existing pressure tests.
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -1095,7 +1110,10 @@ void FileScannerV2::update_realtime_counters() {
_state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_remote_storage(
deltas.scan_bytes_from_remote_storage);
- COUNTER_SET(_file_read_bytes_counter, bytes_read);
+ // Scanner instances share the profile counter, so publishing an absolute
value would erase
+ // bytes already reported by sibling scanners.
+ COUNTER_UPDATE(_file_read_bytes_counter,
Review Comment:
[P2] Aggregate calls and time across sibling scanners too
The sharing rule in this new comment applies to all three adjacent counters,
but only bytes was converted to a per-scanner additive delta. `FileReadCalls`
and `FileReadTime` still use scanner-local cumulative `COUNTER_SET` values here
and in `_collect_profile_before_close()`, so each realtime/final publication
overwrites work already reported by sibling scanners and the result depends on
scheduling/close order. Please track reported calls/time and `COUNTER_UPDATE`
their deltas as well, then add an interleaved two-scanner profile test that
converges to the sum.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -2310,6 +2373,9 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
const size_t idx =
_predicate_indices_by_position_scratch.at(position);
const auto& col = request.predicate_columns[idx];
const auto fid = col.column_id();
+ if (_current_merge_range_reader != nullptr) {
+ RETURN_IF_ERROR(activate_merge_ranges_for_columns({fid}));
Review Comment:
[P1] Activate deferred ranges before skipping later predicates
When an early staged predicate reduces `selected_rows` to zero, this
activation path is never reached for the remaining predicate columns.
`skip_unmaterialized_predicate_columns()` still calls
`NativeColumnReader::skip(batch_rows)` on each of them; inside the current
selected range that executes `read_with_filter()`, but the deferred
`MergeRangeFileReader` still has no registered range and therefore falls
through to direct underlying reads. Highly selective cold remote scans can
regress to one set of unmerged page/header reads for every later predicate
column. Please activate each unmaterialized column before its physical skip (or
retain a logical pending skip until activation), and cover the zero-survivor
staged-predicate path with an underlying-request-count test.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1246,6 +1933,21 @@ Status CachedRemoteFileReader::read_at_impl(size_t
offset, Slice result, size_t*
}
}};
+ const bool bypass_reader_local_cache = io_ctx->bypass_reader_local_cache;
+ MonotonicStopWatch reader_local_probe_watch;
+ reader_local_probe_watch.start();
+ const bool reader_local_hit =
Review Comment:
[P2] Promote full hits in the remote-only policy
The hot probe above cannot help until some path has populated the
reader-local map, but `_read_remote_only_on_cache_miss()` still reads fully
covered blocks via `FileBlock::read()` directly. After the scanner's write
limiter switches to this policy, an unwrapped/large-chunk Parquet reader can
repeatedly hit FileCache on disk without ever creating the 256 KiB promotion,
so the default-enabled cache is ineffective for exactly those revisits. Please
route full covered hits through `_read_local_block()` while retaining
remote-only miss behavior, and test this with an external reader (the current
Doris-table tests disable the cache).
##########
be/src/io/fs/buffered_reader.cpp:
##########
@@ -397,6 +494,39 @@ Status MergeRangeFileReader::_fill_box(int range_index,
size_t start_offset, siz
return Status::OK();
}
+void MergeRangeFileReader::_record_merged_read(int range_index, size_t
start_offset,
+ size_t bytes_read) {
+ if (bytes_read == 0) {
+ return;
+ }
+ if (range_index < 0) {
+ _statistics.merged_useful_bytes += bytes_read;
+ return;
+ }
+ const size_t read_end = start_offset + bytes_read;
+ size_t useful_bytes = 0;
+ size_t future_predicate_bytes = 0;
+ for (size_t index = static_cast<size_t>(range_index);
+ index < _random_access_ranges.size() &&
+ _random_access_ranges[index].start_offset < read_end;
+ ++index) {
+ const auto& range = _random_access_ranges[index];
+ const size_t overlap_start = std::max(start_offset,
range.start_offset);
+ const size_t overlap_end = std::min(read_end, range.end_offset);
+ if (overlap_start >= overlap_end) {
+ continue;
+ }
+ const size_t overlap = overlap_end - overlap_start;
+ useful_bytes += overlap;
+ if (_range_stages[index] > _range_stages[range_index]) {
Review Comment:
[P2] Make future-stage accounting reachable from production
This comparison has no production path with a greater-stage range already
present. Eager readers construct the whole plan with every stage set to 0 and
never call `add_random_access_ranges()`; deferred readers insert only the
current predicate immediately before reading it, and lazy outputs are inserted
together at one stage. `FuturePredicatePrefetchBytes` therefore stays zero,
while the changed nonzero test manually pre-registers stage 1 in a sequence the
scheduler never performs. Please instrument stage metadata/visibility from the
real scheduler (or remove/rename this counter if zero is the intended
invariant) and cover it with a scheduler-level profile test.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
