Gabriel39 commented on code in PR #69779:
URL: https://github.com/apache/doris/pull/69779#discussion_r4438715721
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1206,6 +1207,133 @@ Status
CachedRemoteFileReader::_read_remote_only_on_cache_miss(
return Status::OK();
}
+namespace {
+
+// Bypasses already update the history, and peer IO measures a different
source. Only real
+// remote misses need sampling here, after the caller has checked query
eligibility and success.
+void record_adaptor_remote_sample(bool eligible, const ReadStatistics& stats) {
+ if (eligible && stats.remote_physical_read_bytes > 0 &&
stats.io_adaptor_bypass_count == 0 &&
+ !stats.from_peer_cache) {
+
FileCacheIOAdaptor::instance().record_remote(FileCacheIOAdaptor::latency_per_block_us(
+ static_cast<size_t>(stats.remote_physical_read_bytes),
stats.remote_read_timer,
+ config::file_cache_each_block_size));
+ }
+}
+
+// A partial cache copy cannot be published: the caller retries the complete
range through
+// the normal planner when any block fails, preserving its cache-repair and
remote fallback.
+bool read_adaptor_cached_blocks(BlockFileCache* cache, const UInt128Wrapper&
hash,
+ const FileBlocks& blocks, size_t offset, Slice
result,
+ ReadStatistics& stats) {
+ const auto bytes_req = result.size;
+ const auto local_start = MonotonicNanos();
+ ++stats.io_adaptor_local_reads;
+ Defer local_timer {[&] {
+ const auto elapsed = MonotonicNanos() - local_start;
+ stats.io_adaptor_local_time_ns += elapsed;
+ stats.local_read_timer += elapsed;
+ }};
+ for (const auto& block : blocks) {
+ const auto left = std::max(offset, block->range().left);
+ const auto right = std::min(offset + bytes_req - 1,
block->range().right);
+ Status status = block->read(Slice(result.data + left - offset, right -
left + 1),
+ left - block->range().left);
+ if (!status.ok()) {
+ // Concurrent eviction can remove a previously downloaded file.
The ordinary path
+ // already handles self-healing and remote fallback, so retry the
full request there.
+ if (status.is<ErrorCode::NOT_FOUND>()) {
+ cache->remove_if_cached_async(hash);
+ }
+ return false;
+ }
+ }
+ const auto elapsed = MonotonicNanos() - local_start;
+
FileCacheIOAdaptor::instance().record_local(FileCacheIOAdaptor::latency_per_block_us(
+ bytes_req, elapsed, config::file_cache_each_block_size));
+ return true;
+}
+
+} // namespace
+
+bool CachedRemoteFileReader::_io_adaptor_eligible(size_t bytes, const
IOContext* io_ctx) const {
+ // Only explicitly enabled external query reads may adapt.
Background/index/warmup reads
+ // and memory-cache/inflight payloads retain their existing policies.
After an optional
+ // remote read fails, keep later cache hits on the normal path for this
reader's lifetime.
+ return io_ctx->enable_file_cache_io_adaptor && !_is_doris_table &&
!io_ctx->is_dryrun &&
+ !_io_adaptor_remote_failed.load(std::memory_order_acquire) &&
!io_ctx->is_warmup &&
+ !io_ctx->is_index_data && !io_ctx->is_inverted_index &&
!io_ctx->should_stop &&
+ io_ctx->read_file_cache && bytes > 0 &&
+ _cache->get_storage()->get_type() == FileCacheStorageType::DISK;
+}
+
+Status CachedRemoteFileReader::_try_read_with_io_adaptor(
+ size_t offset, Slice result, size_t bytes_req, size_t* bytes_read,
const IOContext* io_ctx,
+ ReadStatistics& stats, SourceReadBreakdown& sources, bool* handled) {
+ *handled = false;
+ FileBlocks blocks;
+ bool fully_covered = false;
+ CacheContext context(io_ctx);
+ context.stats = &stats;
+ context.tablet_id = _tablet_id;
+ ++stats.io_adaptor_lookup_count;
+ {
+ SCOPED_RAW_TIMER(&stats.io_adaptor_lookup_time_ns);
+ RETURN_IF_ERROR(_cache->get_downloaded_blocks_if_fully_covered(
+ _cache_hash, offset, bytes_req, context, &blocks,
&fully_covered));
+ }
+ if (!fully_covered) {
+ // A mixed hit/miss or inflight-buffer request follows the existing
sync/async planner.
+ // In particular, a bypass must never acquire downloader ownership or
cause writeback.
+ return Status::OK();
+ }
+ auto& adaptor = FileCacheIOAdaptor::instance();
+ const auto decision =
adaptor.select(config::file_cache_io_adaptor_skip_read_factor);
+ stats.io_adaptor_local_feedbacks += decision.local_feedback;
+ stats.io_adaptor_remote_feedbacks += decision.remote_feedback;
+ if (decision.read_remote()) {
+ // Requested and returned bytes differ for failed/short reads
recovered from local cache.
+ ++stats.io_adaptor_bypass_count;
+ stats.io_adaptor_bypass_bytes += bytes_req;
+ size_t remote_bytes = 0;
+ s3_read_counter << 1;
+ const auto start = MonotonicNanos();
+ Status status = _remote_file_reader->read_at(offset,
Slice(result.data, bytes_req),
+ &remote_bytes, io_ctx);
+ const auto elapsed = MonotonicNanos() - start;
+ stats.io_adaptor_remote_time_ns += elapsed;
+ stats.remote_read_timer += elapsed;
+ stats.io_adaptor_remote_bytes += remote_bytes;
+ if (status.ok() && remote_bytes == bytes_req) {
+ stats.remote_physical_read_bytes += remote_bytes;
+ adaptor.record_remote(FileCacheIOAdaptor::latency_per_block_us(
+ remote_bytes, elapsed,
config::file_cache_each_block_size));
+ sources.remote_bytes += remote_bytes;
+ stats.hit_cache = false;
+ stats.skip_cache = true;
+ *bytes_read = remote_bytes;
+ *handled = true;
+ return Status::OK();
+ }
+ ++stats.io_adaptor_remote_failures;
+ if (io_ctx->should_stop || status.is<ErrorCode::CANCELLED>()) {
+ return status.ok() ? Status::Cancelled("Adaptive cache read
cancelled") : status;
+ }
+ // Repeating an optional failing GET would delay every subsequent
cached range. Keep
+ // required remote misses available without poisoning other readers'
shared estimates.
+ _io_adaptor_remote_failed.store(true, std::memory_order_release);
Review Comment:
Fixed in efd0eed115. HDFS readers retain the established cache path because
an optional HDFS read error can destroy the backing handle. A real HDFS-backed
reader test seeds cached bytes and verifies repeated cache hits preserve both
the unopened handle and file size. Required misses retain the original reader
behavior.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1276,6 +1406,14 @@ Status CachedRemoteFileReader::read_at_impl(size_t
offset, Slice result, size_t*
}
}};
+ if (_io_adaptor_eligible(bytes_req, io_ctx)) {
+ bool handled = false;
+ read_st = _try_read_with_io_adaptor(offset, result, bytes_req,
bytes_read, io_ctx, stats,
+ source_read_breakdown, &handled);
+ if (!read_st.ok() || handled) {
+ return read_st;
Review Comment:
Fixed in efd0eed115. Completed adaptor local hits now update all six
established per-disk read/hit counters and byte totals using complete
cache-block sizes. Tests cover both write modes, remote bypasses, incomplete
lookups and failed local-copy retries without duplicate accounting. The
existing external cache-statistics regression remains unchanged.
--
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]
