Gabriel39 commented on code in PR #69768:
URL: https://github.com/apache/doris/pull/69768#discussion_r4438736720
##########
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:
Synced the fix from #69779 in d0e5271507. HDFS stays on the original cache
path because failed optional reads can destroy its backing handle. The
HDFS-backed reader regression verifies repeated cached reads preserve the
unopened handle and file size.
##########
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:
Synced the fix from #69779 in d0e5271507. Successful adaptor local hits now
update the six established per-disk read/hit counters and byte totals. Both
write modes are covered; remote bypasses, incomplete lookups and failed
disk-copy retries do not duplicate accounting.
##########
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));
Review Comment:
Fixed in d0e5271507 and synchronized with #69779. Full inflight coverage is
checked before the adaptor acquires the disk-cache metadata mutex. The existing
async planner retains materialization and counter ownership. The enabled
external-reader regression now holds that mutex while requiring the memory-only
read to finish; it reproduced the stall before the fix and passes afterward.
--
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]
