github-actions[bot] commented on code in PR #69768: URL: https://github.com/apache/doris/pull/69768#discussion_r4400290934
########## be/src/io/cache/file_cache_io_adaptor.cpp: ########## @@ -0,0 +1,114 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "io/cache/file_cache_io_adaptor.h" + +#include <algorithm> + +namespace doris::io { + +FileCacheIOAdaptor& FileCacheIOAdaptor::instance() { + static FileCacheIOAdaptor adaptor; + return adaptor; +} + +void FileCacheIOAdaptor::Window::append(int64_t latency_us) { + if (count == WINDOW_SIZE) { + sum -= samples[next]; + } else { + ++count; + } + samples[next] = latency_us; + sum += latency_us; + next = (next + 1) % WINDOW_SIZE; +} + +void FileCacheIOAdaptor::Window::record(int64_t latency_us) { + if (latency_us < 0) { + return; + } + if (latency_us < minimum || minimum_life > MINIMUM_LIFE_LIMIT) { + minimum = latency_us; + minimum_life = 0; + } else { + ++minimum_life; + } + append(latency_us); +} + +bool FileCacheIOAdaptor::Window::feedback() { + if (minimum < 0 || minimum > UPPER_MIN_LATENCY_US) { + return false; + } + // Feed the remembered minimum into the unused path to reconsider its estimated cost. + // Only actual observations may advance the minimum's observation age. + append(minimum); + return true; +} + +int64_t FileCacheIOAdaptor::Window::average() const { + return count == 0 ? 0 : static_cast<int64_t>(sum / count); +} + +FileCacheIOAdaptor::Decision FileCacheIOAdaptor::select(double skip_read_factor) { + std::lock_guard lock(_mutex); + Decision decision; + decision.local_latency_us = _local.average(); + decision.remote_latency_us = _remote.average(); + if (_local.count < WINDOW_SIZE || _remote.count < WINDOW_SIZE || + decision.local_latency_us < MIN_LOCAL_LATENCY_US) { + return decision; + } + // Preserve fractional factors when comparing the two average service times. + decision.bypass = static_cast<double>(decision.local_latency_us) * skip_read_factor > + static_cast<double>(decision.remote_latency_us); + if (decision.bypass) { + decision.local_feedback = _local.feedback(); Review Comment: [P2] Periodically observe the local path while bypassing it If the first 50 local observations occur during congestion (e.g. 20ms versus remote 1ms), the remembered local minimum is also 20ms. Once bypass starts, `_local.feedback()` only appends that same value, while successful bypasses never perform a real local read. If the disk later recovers to sub-millisecond latency, every subsequent fully cached read still goes remote indefinitely; `minimum_life` cannot expire the estimate because it advances only on actual local observations. The recovery test seeds an earlier fast local minimum and misses this reachable case. Add bounded real probes or another exploration mechanism that can discover recovery without a historical fast sample. ########## be/src/io/cache/cached_remote_file_reader.cpp: ########## @@ -1197,6 +1198,101 @@ 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. + return io_ctx->enable_file_cache_io_adaptor && !_is_doris_table && !io_ctx->is_dryrun && + !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; Review Comment: [P1] Stop selecting a failing remote path after cache fallback Only complete successful bypass reads call `record_remote()`. After 50 local=20ms/remote=1ms samples, an S3 outage or repeated short reads leave the remote estimate at 1ms forever: each fully cached request waits through the failing remote GET/retries, falls back to the 20ms cached read, then makes the same bypass decision again. The outer sampler also excludes any request with `io_adaptor_bypass_count != 0`, so nothing corrects the estimate. This can turn a usable warm cache into repeated network timeouts. Feed failures into routing/backoff so subsequent cached reads avoid the failing remote path. ########## be/src/io/cache/cached_remote_file_reader.cpp: ########## @@ -1197,6 +1198,101 @@ 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. + return io_ctx->enable_file_cache_io_adaptor && !_is_doris_table && !io_ctx->is_dryrun && + !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( Review Comment: [P2] Preserve asynchronous LRU touches on the new cache-hit path With `enable_file_cache_async_touch_on_get_or_set=true`, this lookup calls `use_cell()`, which skips synchronous LRU movement, but `get_downloaded_blocks_if_fully_covered()` never enqueues an asynchronous touch. Returning handled from the adaptor then skips both the sync path's `get_or_set()` touch enqueue and the async path's `touch_probe_block_if_cached()`. Thus even local hits before the controller has enough samples stop refreshing recency: with oldest A then newer B, repeatedly reading A and inserting C at capacity evicts A before cold B. Updating atime does not help same-queue capacity eviction. Preserve the existing touch behavior and cover this setting in the reader tests. ########## gensrc/thrift/PaloInternalService.thrift: ########## @@ -518,6 +518,10 @@ struct TQueryOptions { // Push LIMIT into SegmentIterator when safe. 231: optional bool enable_segment_limit_pushdown = true + + // Explicitly sent by new FEs; an omitted field keeps older-FE queries on the local policy. + 232: optional bool enable_file_cache_io_adaptor = false; Review Comment: [P1] Allocate a query-option ID that is unused across branches Field 232 is already `enable_runtime_filter_bucket_prune` (also BOOL) on current master (`62e3d81a42b2b74ee2a5a577969df62edee06b78`). Master FE explicitly sends it in `SessionVariable.toThrift()`, so a mixed-version master-FE/4.2-BE query will interpret bucket-pruning=true as an explicitly enabled cache adaptor; conversely, a 4.2 FE setting adaptor=false disables bucket pruning on an upgraded BE. Thrift cannot distinguish the meanings because both use the same ID and wire type. Please allocate/reserve one globally unused ID consistently across release branches rather than reusing 232. -- 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]
