github-actions[bot] commented on code in PR #66504:
URL: https://github.com/apache/doris/pull/66504#discussion_r3746335429
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1050,6 +1111,98 @@ Status
CachedRemoteFileReader::_read_remaining_blocks_from_cache(
return Status::OK();
}
+Status CachedRemoteFileReader::read_at_from_cache(size_t offset, Slice result,
size_t* bytes_read,
+ bool* cache_hit, const
IOContext* io_ctx) {
+ IOContext default_io_ctx;
+ if (io_ctx == nullptr) {
+ io_ctx = &default_io_ctx;
+ }
+ const auto read_type =
+ io_ctx->is_inverted_index
+ ? FileCacheReadType::INVERTED_INDEX
+ : (io_ctx->is_index_data ?
FileCacheReadType::SEGMENT_FOOTER_INDEX
+ : FileCacheReadType::DATA);
+ auto publish_stats = [&](const ReadStatistics& stats,
+ const SourceReadBreakdown& source_read_breakdown)
{
+ if (io_ctx->is_dryrun) {
+ return;
+ }
+ if (io_ctx->file_cache_stats != nullptr) {
+ _update_stats(stats, source_read_breakdown,
io_ctx->file_cache_stats, read_type);
+ }
+ if (!io_ctx->is_warmup) {
+ FileCacheStatistics increment;
+ _update_stats(stats, source_read_breakdown, &increment, read_type);
+ FileCacheMetrics::instance().update(&increment);
+ }
+ };
+ *bytes_read = 0;
+ *cache_hit = false;
+ const size_t bytes_req = std::min(result.size, size() - std::min(offset,
size()));
+ if (bytes_req == 0) {
+ *cache_hit = true;
+ return Status::OK();
+ }
+ if (_read_from_memory_block_cache(offset, Slice(result.data, bytes_req))) {
+ *bytes_read = bytes_req;
+ *cache_hit = true;
+ ReadStatistics stats;
+ stats.bytes_read = cast_set<int64_t>(bytes_req);
+ SourceReadBreakdown source_read_breakdown;
+ source_read_breakdown.local_bytes = cast_set<int64_t>(bytes_req);
+ publish_stats(stats, source_read_breakdown);
+ return Status::OK();
+ }
+
+ ReadStatistics stats;
+ SourceReadBreakdown source_read_breakdown;
+ stats.bytes_read = cast_set<int64_t>(bytes_req);
+ const size_t block_size =
cast_set<size_t>(config::file_cache_each_block_size);
+ const size_t align_left = offset / block_size * block_size;
+ const size_t request_end = offset + bytes_req;
+ const size_t align_end =
+ std::min((request_end + block_size - 1) / block_size * block_size,
size());
+ const size_t align_size = align_end - align_left;
+ CacheContext cache_context(io_ctx);
+ cache_context.stats = &stats;
+ MonotonicStopWatch sw;
+ sw.start();
+ FileBlocksHolder holder =
+ _cache->get_or_set(_cache_hash, align_left, align_size,
cache_context);
+ stats.cache_get_or_set_timer += sw.elapsed_time();
+ if (std::ranges::any_of(holder.file_blocks, [](const auto& block) {
Review Comment:
[P2] Avoid repeating a mutating cache lookup on every miss
This `get_or_set()` is not a read-only coverage probe: it can create/touch
cache cells and updates BlockFileCache lookup/hit metrics. Returning here drops
its query-side timers, after which MergeRange falls back through
`CachedRemoteFileReader::read_at()` and performs the same `get_or_set()` again.
Partial coverage and eviction-after-copy take the same split-accounting path.
Please use a non-mutating exact-coverage lookup or carry this holder/result
into fallback so one logical read performs one cache lookup and publishes its
costs once.
##########
be/src/io/fs/buffered_reader.cpp:
##########
@@ -45,6 +45,38 @@ namespace doris {
namespace io {
struct IOContext;
+Status MergeRangeFileReader::add_random_access_ranges(const
std::vector<PrefetchRange>& ranges,
+ uint32_t stage) {
+ for (const auto& range : ranges) {
+ if (range.start_offset >= range.end_offset) {
+ return Status::InvalidArgument("Invalid merge-read range [{},
{})", range.start_offset,
+ range.end_offset);
+ }
+ auto it = std::lower_bound(_random_access_ranges.begin(),
_random_access_ranges.end(),
+ range.start_offset,
+ [](const PrefetchRange& lhs, size_t
start_offset) {
+ return lhs.start_offset < start_offset;
+ });
+ const size_t index = static_cast<size_t>(it -
_random_access_ranges.begin());
+ if (it != _random_access_ranges.end() && it->start_offset ==
range.start_offset) {
+ if (it->end_offset != range.end_offset) {
+ return Status::InvalidArgument("Overlapping merge-read
ranges");
+ }
+ _range_stages[index] = std::min(_range_stages[index], stage);
+ continue;
+ }
+ if ((it != _random_access_ranges.begin() &&
Review Comment:
[P1] Accept compatible PARQUET-816 padded overlaps
These ranges come from `compute_column_chunk_range(...,
parquet_816_padding=true)`, which deliberately extends every affected pre-1.2.9
parquet-mr chunk by up to 100 bytes. Adjacent chunks can therefore have valid
overlapping extents. On a remote/exact-cache scan with two predicate leaves,
activating the second range reaches this branch and returns `InvalidArgument`,
aborting a file that the native reader otherwise supports. Please
normalize/coalesce these compatibility ranges (while preserving their
per-column cursor contract) instead of rejecting the overlap, and cover an
old-writer adjacent-chunk case.
##########
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 required benchmark guide in sync
This seventh selectivity changes the decoder registration count from 228 to
266, as the updated invariant test confirms. However
`be/benchmark/parquet/AGENTS.md` still tells maintainers to expect 228 and
still describes only 0/1/10/50/90/100 at lines 51, 137-138, and 353. Please
update those required build/list/coverage instructions together with this loop.
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -1095,7 +1103,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,
+ _cumulative_profile_delta(bytes_read,
&_reported_file_read_bytes));
COUNTER_SET(_file_read_calls_counter,
cast_set<int64_t>(_file_reader_stats->read_calls));
Review Comment:
[P2] Accumulate calls and time across sibling scanners too
The comment above applies to all three counters: every FileScannerV2 created
by one FileScanLocalState registers against the same scanner profile, and
`add_counter()` returns the existing object. If scanner A publishes 100
calls/10 ms and scanner B then publishes 40 calls/4 ms, these SETs leave 40/4
rather than 140/14 (and close repeats the overwrite). Track per-scanner
reported calls/time and UPDATE their deltas in both publication paths, as this
patch now does for bytes.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -800,6 +800,67 @@ bool
CachedRemoteFileReader::_try_read_from_cached_files_directly(
return false;
}
+bool CachedRemoteFileReader::_read_from_memory_block_cache(size_t offset,
Slice result) {
+ std::lock_guard lock(_memory_block_cache_mutex);
+ auto it = _memory_block_cache.upper_bound(offset);
+ if (it == _memory_block_cache.begin()) {
+ return false;
+ }
+ --it;
+ const size_t relative_offset = offset - it->first;
+ if (relative_offset > it->second->size() ||
+ result.size > it->second->size() - relative_offset) {
+ return false;
+ }
+ memcpy(result.data, it->second->data() + relative_offset, result.size);
+ return true;
+}
+
+Status CachedRemoteFileReader::_read_local_block(const FileBlockSPtr& block,
size_t file_offset,
+ size_t absolute_offset, Slice
result) {
+ if (_read_from_memory_block_cache(absolute_offset, result)) {
+ return Status::OK();
+ }
+ if (_is_doris_table) {
+ return block->read(result, file_offset);
+ }
+
+ bool promote = false;
+ {
+ std::lock_guard lock(_memory_block_cache_mutex);
+ auto& accesses = _memory_block_access_counts[block->offset()];
+ accesses = std::min<uint8_t>(accesses + 1,
MEMORY_BLOCK_PROMOTION_ACCESSES);
+ promote = accesses == MEMORY_BLOCK_PROMOTION_ACCESSES;
+ if (_memory_block_access_counts.size() > 128) {
+
_memory_block_access_counts.erase(_memory_block_access_counts.begin());
+ }
+ }
+ if (!promote || block->range().size() > MAX_MEMORY_BLOCK_CACHE_BYTES) {
+ return block->read(result, file_offset);
+ }
+
+ auto block_data =
std::make_shared<std::vector<char>>(block->range().size());
Review Comment:
[P1] Budget and account the promoted block copies globally
A third small page access now reads and retains the entire FileCache block,
but both callers attribute only the requested slice. The retained map is capped
at 16 MiB per reader, not per query/backend; the default 16 file scanners in
one local state can therefore hold another 256 MiB of duplicated FileCache
data, with no pressure-driven cache eviction or retained/promotion profile.
Please put promotion under an aggregate cache/query budget (or reuse a shared
bounded cache), expose the full-block read/retained bytes, and add multi-reader
pressure coverage.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -2310,6 +2368,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) {
Review Comment:
[P1] Expose ranges before skipping later predicate readers
This activates a column only when it is materialized. If an earlier staged
predicate reduces the batch to zero, the early-exit path instead calls
`skip_unmaterialized_predicate_columns()` for every later reader without
exposing those ranges. `NativeColumnReader::skip()` still drives
`read_with_filter(..., filter_all=true)` over selected rows, so the empty
MergeRange plan delegates the page reads directly and loses
coalescing/exact-cache handling on the most selective remote path. Activate the
skipped columns as a stage before that loop (or provide a genuinely I/O-free
skip contract), and add a scheduler-level all-filtered 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]
