github-actions[bot] commented on code in PR #69768:
URL: https://github.com/apache/doris/pull/69768#discussion_r4402390989
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1197,6 +1198,106 @@ Status
CachedRemoteFileReader::_read_remote_only_on_cache_miss(
return Status::OK();
}
+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);
+ // The same immutable range is still cached. Overwrite the complete
caller range after
+ // an error/short remote read; never publish the partially overwritten
buffer as success.
+ ++stats.io_adaptor_fallback_local;
+ }
+ 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(_cache_hash);
+ }
+ return Status::OK();
+ }
+ }
+ const auto elapsed = MonotonicNanos() - local_start;
+ adaptor.record_local(FileCacheIOAdaptor::latency_per_block_us(
+ bytes_req, elapsed, config::file_cache_each_block_size));
+ sources.local_bytes += bytes_req;
+ *bytes_read = bytes_req;
+ *handled = true;
Review Comment:
[P2] Keep cache-level hit accounting on adaptor local reads
With default synchronous writes, read a cold full block once and then read
it 99 more times with the adaptor enabled and insufficient remote history to
bypass. Those 99 local full hits return `handled=true` here and skip
`get_or_set()`, the only updater of `_num_read_blocks`, `_num_hit_blocks`,
`_total_read_size_metrics`, `_total_hit_size_metrics` and the no-warmup
counters. `get_downloaded_blocks_if_fully_covered()` only pins/touches cells,
so the metrics remain at 0 hits/1 read instead of 99/100; existing hit-ratio
gauges and cache statistics report a cold cache while reads hit disk cache. The
new query Profile counters do not update these cache-level metrics. Preserve
this accounting on completed adaptor hits without double counting planner
fallbacks.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -1197,6 +1198,106 @@ Status
CachedRemoteFileReader::_read_remote_only_on_cache_miss(
return Status::OK();
}
+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:
[P1] Keep the HDFS reader usable after a failed optional read
This policy also applies to cached HDFS files (`FileFactory` wraps
`HdfsFileReader`). If `hdfsPread`/`hdfsRead` returns an error during a bypass,
`HdfsFileReader::read_at_impl()` clears `_handle` and destroys its accessor.
The local fallback below returns success, so an ordinary multi-range scan
continues with that same reader; its next `read_at_impl()` calls `size()`
before checking `_io_adaptor_remote_failed`, and `HdfsFileReader::size()`
dereferences the null handle. A transient HDFS error on a fully cached file can
therefore crash the BE. Preserve stable size metadata and a usable/reopenable
HDFS reader across fallback, or exclude readers with destructive failure
semantics; cover a failed bypass followed by another cached read using HDFS
failure semantics.
--
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]
