This is an automated email from the ASF dual-hosted git repository. Gabriel39 pushed a commit to branch 4.1.5-iceberg-opt in repository https://gitbox.apache.org/repos/asf/doris.git
commit 7af14bb9d998cb1188dec1ab93552330e6553e90 Author: Gabriel <[email protected]> AuthorDate: Wed Oct 28 14:35:59 2026 +0800 [improvement](file-cache) adapt external cache reads to I/O latency A warm external scan can stall on a congested cache disk even when object storage serves the same immutable bytes faster. Add a shared BE-wide latency controller aligned with the external-read policy in StarCache 4.1.4: retain 50 samples per path, require both histories, keep reading locally when its average is below 5 ms, and bypass when local average * skip factor exceeds remote average. The unselected path contributes its remembered minimum to its sliding window; only actual observations age that minimum. Do not add expiry, periodic real probes, resource/size shards, request-size caps or extra concurrency limits. Expose enable_file_cache_io_adaptor as a forwarded session variable, default true, and retain only file_cache_io_adaptor_skip_read_factor=1.0 on the BE. Explicit optional Thrift propagation keeps older FE queries on the original policy. Background, LOAD, index, warmup, memory-cache and internal-table reads remain outside this external-query policy. Adapt complete downloaded disk-cache hits before sync/async dispatch. Mixed misses and inflight buffers retain their existing planners. Direct bypass retains cached blocks and never takes downloader ownership or rewrites them. Remote errors and short reads overwrite the complete range from local cache; cancellation propagates, and missing cache files retain normal self-healing. Successful physical remote misses train the same shared history in both modes. Normalize service times to Doris's native cache block using the empirical ratio of two. Also normalize partial local reads because they lack SR's block buffer. Keep fractional factors intact instead of the reference library's integer truncation. These integration differences are intentional; the per-block mixed-range planner and memory caching are unchanged. Add 12 Profile counters for lookup cost, local reads/time, attempted bypass count/bytes, actual remote bytes/time, minimum feedback on both paths, remote failures and local fallback. Cover statistics merge, delta and final Profile reporting so repeated reports do not double-count. Measured effect and validation: - Five alternating on/off rounds, 200 x 64 KiB requests per group, after calibration with 50 actual service-time samples on each path. File-backed remote reads and injected delays make this a controlled local experiment. - Local 8 ms / remote 1 ms: median 1,711.451 ms disabled versus 236.368 ms enabled (7.24x speedup, 86.19% latency reduction). Each enabled group bypasses 200 requests / 13,107,200 bytes and contributes 200 local minimum-feedback samples without issuing extra probe reads. - Local 1 ms / remote 8 ms: median 219.364 ms disabled versus 218.181 ms enabled (-0.54%), with zero remote bypasses. - Both scenarios preserve every returned byte and perform zero cache writes. Production query gains and end-to-end runtime-filter timing need separate measurement; this benchmark does not establish those gains. - Release BE build and GLIBC 2.17 compatibility check passed. - 46 focused BE tests passed: 15 controller, 10 reader integration, 13 existing async-reader, 5 statistics/Profile and 3 shared scanner IO-context tests. Both opt-in cache-path benchmarks also passed. - All 15 controller tests passed with ASan/UBSan and leak detection enabled. - Two FE session serialization/isolation/forwarding tests passed; full FE Checkstyle and clang-format 16 checks passed. - Against the actual packaged StarCache 4.1.4 policy object, factors 1, 2 and 3 each matched all 43,018 decisions in 600,157 identical input operations. --- be/src/common/config.cpp | 6 + be/src/common/config.h | 2 + be/src/exec/scan/file_scan_io_context.h | 4 + be/src/io/cache/block_file_cache_profile.cpp | 48 ++ be/src/io/cache/block_file_cache_profile.h | 12 + be/src/io/cache/cached_remote_file_reader.cpp | 130 +++++- be/src/io/cache/cached_remote_file_reader.h | 6 + be/src/io/cache/file_block.cpp | 1 + be/src/io/cache/file_cache_common.h | 13 + be/src/io/cache/file_cache_io_adaptor.cpp | 114 +++++ be/src/io/cache/file_cache_io_adaptor.h | 72 +++ be/src/io/io_common.h | 26 ++ be/test/exec/scan/file_scanner_v2_test.cpp | 26 ++ .../block_file_cache_profile_reporter_test.cpp | 48 ++ .../io/cache/cached_remote_file_reader_test.cpp | 484 +++++++++++++++++++++ be/test/io/cache/file_cache_io_adaptor_test.cpp | 198 +++++++++ .../java/org/apache/doris/qe/SessionVariable.java | 7 + .../qe/SessionVariableFileCacheIoAdaptorTest.java | 51 +++ gensrc/thrift/PaloInternalService.thrift | 3 + 19 files changed, 1250 insertions(+), 1 deletion(-) diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 0e34b88d708..75c9083d52b 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -25,6 +25,7 @@ #include <lz4/lz4hc.h> #include <cerrno> // IWYU pragma: keep +#include <cmath> #include <cstdlib> #include <cstring> #include <fstream> // IWYU pragma: keep @@ -1322,6 +1323,11 @@ DEFINE_mBool(file_cache_enable_only_warm_up_idx, "false"); DEFINE_Int32(file_cache_downloader_thread_num_min, "32"); DEFINE_Int32(file_cache_downloader_thread_num_max, "32"); +// Larger values favor remote reads when a query enables the file-cache IO adaptor. +DEFINE_mDouble(file_cache_io_adaptor_skip_read_factor, "1.0"); +DEFINE_Validator(file_cache_io_adaptor_skip_read_factor, + [](double v) { return std::isfinite(v) && v > 0; }); + // async file cache write DEFINE_mBool(enable_async_file_cache_write, "false"); DEFINE_mInt32(async_file_cache_write_workers_per_disk, "16"); diff --git a/be/src/common/config.h b/be/src/common/config.h index 3931b11efed..96f252bcb8f 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1349,6 +1349,8 @@ DECLARE_mBool(enable_evaluate_shadow_queue_diff); DECLARE_mBool(file_cache_enable_only_warm_up_idx); +DECLARE_mDouble(file_cache_io_adaptor_skip_read_factor); + // async file cache write DECLARE_mBool(enable_async_file_cache_write); DECLARE_mInt32(async_file_cache_write_workers_per_disk); diff --git a/be/src/exec/scan/file_scan_io_context.h b/be/src/exec/scan/file_scan_io_context.h index 85bd6ae1abd..8e91bbb9812 100644 --- a/be/src/exec/scan/file_scan_io_context.h +++ b/be/src/exec/scan/file_scan_io_context.h @@ -29,6 +29,10 @@ inline void init_file_scan_io_context(RuntimeState* state, io::IOContext* io_ctx io_ctx->query_id = &state->query_id(); if (state->query_options().query_type == TQueryType::SELECT) { io_ctx->reader_type = ReaderType::READER_QUERY; + // Missing options from an older FE must not enable a new remote-read policy. + io_ctx->enable_file_cache_io_adaptor = + state->query_options().__isset.enable_file_cache_io_adaptor && + state->query_options().enable_file_cache_io_adaptor; if (auto* query_ctx = state->get_query_ctx(); query_ctx != nullptr) { io_ctx->remote_scan_cache_write_limiter = query_ctx->remote_scan_cache_write_limiter(); } diff --git a/be/src/io/cache/block_file_cache_profile.cpp b/be/src/io/cache/block_file_cache_profile.cpp index 366cf6fd684..f7b9802f7ca 100644 --- a/be/src/io/cache/block_file_cache_profile.cpp +++ b/be/src/io/cache/block_file_cache_profile.cpp @@ -99,6 +99,18 @@ FileCacheStatistics diff_file_cache_statistics(const FileCacheStatistics& curren SUBTRACT_FIELD(lock_wait_timer); SUBTRACT_FIELD(get_timer); SUBTRACT_FIELD(set_timer); + SUBTRACT_FIELD(io_adaptor_lookup_count); + SUBTRACT_FIELD(io_adaptor_lookup_time_ns); + SUBTRACT_FIELD(io_adaptor_local_reads); + SUBTRACT_FIELD(io_adaptor_local_time_ns); + SUBTRACT_FIELD(io_adaptor_bypass_count); + SUBTRACT_FIELD(io_adaptor_bypass_bytes); + SUBTRACT_FIELD(io_adaptor_remote_bytes); + SUBTRACT_FIELD(io_adaptor_remote_time_ns); + SUBTRACT_FIELD(io_adaptor_local_feedbacks); + SUBTRACT_FIELD(io_adaptor_remote_feedbacks); + SUBTRACT_FIELD(io_adaptor_remote_failures); + SUBTRACT_FIELD(io_adaptor_fallback_local); SUBTRACT_FIELD(async_cache_write_submitted); SUBTRACT_FIELD(async_cache_write_rejected); SUBTRACT_FIELD(async_cache_write_buffer_alloc_fail); @@ -176,6 +188,30 @@ FileCacheProfileReporter::FileCacheProfileReporter(RuntimeProfile* profile, lock_wait_timer = ADD_CHILD_TIMER_WITH_LEVEL(profile, "LockWaitTimer", cache_profile, 1); get_timer = ADD_CHILD_TIMER_WITH_LEVEL(profile, "GetTimer", cache_profile, 1); set_timer = ADD_CHILD_TIMER_WITH_LEVEL(profile, "SetTimer", cache_profile, 1); + io_adaptor_lookup_count = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLookupCount", + TUnit::UNIT, cache_profile, 1); + io_adaptor_lookup_time_ns = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLookupTime", + TUnit::TIME_NS, cache_profile, 1); + io_adaptor_local_reads = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLocalReadCount", + TUnit::UNIT, cache_profile, 1); + io_adaptor_local_time_ns = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLocalReadTime", + TUnit::TIME_NS, cache_profile, 1); + io_adaptor_bypass_count = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorBypassCount", + TUnit::UNIT, cache_profile, 1); + io_adaptor_bypass_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorBypassBytes", + TUnit::BYTES, cache_profile, 1); + io_adaptor_remote_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorRemoteReadBytes", + TUnit::BYTES, cache_profile, 1); + io_adaptor_remote_time_ns = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorRemoteReadTime", + TUnit::TIME_NS, cache_profile, 1); + io_adaptor_local_feedbacks = ADD_CHILD_COUNTER_WITH_LEVEL( + profile, "IOAdaptorLocalFeedbackCount", TUnit::UNIT, cache_profile, 1); + io_adaptor_remote_feedbacks = ADD_CHILD_COUNTER_WITH_LEVEL( + profile, "IOAdaptorRemoteFeedbackCount", TUnit::UNIT, cache_profile, 1); + io_adaptor_remote_failures = ADD_CHILD_COUNTER_WITH_LEVEL( + profile, "IOAdaptorRemoteFailureCount", TUnit::UNIT, cache_profile, 1); + io_adaptor_fallback_local = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorFallbackLocalCount", + TUnit::UNIT, cache_profile, 1); async_cache_write_submitted = ADD_CHILD_COUNTER_WITH_LEVEL( profile, "AsyncCacheWriteSubmittedCount", TUnit::UNIT, cache_profile, 1); async_cache_write_rejected = ADD_CHILD_COUNTER_WITH_LEVEL( @@ -298,6 +334,18 @@ void FileCacheProfileReporter::update(const FileCacheStatistics* statistics) con COUNTER_UPDATE(lock_wait_timer, statistics->lock_wait_timer); COUNTER_UPDATE(get_timer, statistics->get_timer); COUNTER_UPDATE(set_timer, statistics->set_timer); + COUNTER_UPDATE(io_adaptor_lookup_count, statistics->io_adaptor_lookup_count); + COUNTER_UPDATE(io_adaptor_lookup_time_ns, statistics->io_adaptor_lookup_time_ns); + COUNTER_UPDATE(io_adaptor_local_reads, statistics->io_adaptor_local_reads); + COUNTER_UPDATE(io_adaptor_local_time_ns, statistics->io_adaptor_local_time_ns); + COUNTER_UPDATE(io_adaptor_bypass_count, statistics->io_adaptor_bypass_count); + COUNTER_UPDATE(io_adaptor_bypass_bytes, statistics->io_adaptor_bypass_bytes); + COUNTER_UPDATE(io_adaptor_remote_bytes, statistics->io_adaptor_remote_bytes); + COUNTER_UPDATE(io_adaptor_remote_time_ns, statistics->io_adaptor_remote_time_ns); + COUNTER_UPDATE(io_adaptor_local_feedbacks, statistics->io_adaptor_local_feedbacks); + COUNTER_UPDATE(io_adaptor_remote_feedbacks, statistics->io_adaptor_remote_feedbacks); + COUNTER_UPDATE(io_adaptor_remote_failures, statistics->io_adaptor_remote_failures); + COUNTER_UPDATE(io_adaptor_fallback_local, statistics->io_adaptor_fallback_local); COUNTER_UPDATE(async_cache_write_submitted, statistics->async_cache_write_submitted); COUNTER_UPDATE(async_cache_write_rejected, statistics->async_cache_write_rejected); COUNTER_UPDATE(async_cache_write_buffer_alloc_fail, diff --git a/be/src/io/cache/block_file_cache_profile.h b/be/src/io/cache/block_file_cache_profile.h index 4ffe8f0835e..72918fa759d 100644 --- a/be/src/io/cache/block_file_cache_profile.h +++ b/be/src/io/cache/block_file_cache_profile.h @@ -90,6 +90,18 @@ struct FileCacheProfileReporter { RuntimeProfile::Counter* lock_wait_timer = nullptr; RuntimeProfile::Counter* get_timer = nullptr; RuntimeProfile::Counter* set_timer = nullptr; + RuntimeProfile::Counter* io_adaptor_lookup_count = nullptr; + RuntimeProfile::Counter* io_adaptor_lookup_time_ns = nullptr; + RuntimeProfile::Counter* io_adaptor_local_reads = nullptr; + RuntimeProfile::Counter* io_adaptor_local_time_ns = nullptr; + RuntimeProfile::Counter* io_adaptor_bypass_count = nullptr; + RuntimeProfile::Counter* io_adaptor_bypass_bytes = nullptr; + RuntimeProfile::Counter* io_adaptor_remote_bytes = nullptr; + RuntimeProfile::Counter* io_adaptor_remote_time_ns = nullptr; + RuntimeProfile::Counter* io_adaptor_local_feedbacks = nullptr; + RuntimeProfile::Counter* io_adaptor_remote_feedbacks = nullptr; + RuntimeProfile::Counter* io_adaptor_remote_failures = nullptr; + RuntimeProfile::Counter* io_adaptor_fallback_local = nullptr; RuntimeProfile::Counter* async_cache_write_submitted = nullptr; RuntimeProfile::Counter* async_cache_write_rejected = nullptr; RuntimeProfile::Counter* async_cache_write_buffer_alloc_fail = nullptr; diff --git a/be/src/io/cache/cached_remote_file_reader.cpp b/be/src/io/cache/cached_remote_file_reader.cpp index 0d0a0dbf3cb..ec9f7ba0457 100644 --- a/be/src/io/cache/cached_remote_file_reader.cpp +++ b/be/src/io/cache/cached_remote_file_reader.cpp @@ -48,6 +48,7 @@ #include "io/cache/block_file_cache_factory.h" #include "io/cache/block_file_cache_profile.h" #include "io/cache/file_block.h" +#include "io/cache/file_cache_io_adaptor.h" #include "io/cache/peer_file_cache_reader.h" #include "io/cache/remote_scan_cache_write_limiter.h" #include "io/fs/file_reader.h" @@ -328,7 +329,11 @@ Status execute_s3_read(size_t empty_start, size_t& size, std::unique_ptr<char[]> s3_read_counter << 1; SCOPED_RAW_TIMER(&stats.remote_read_timer); stats.from_peer_cache = false; - return remote_file_reader->read_at(empty_start, Slice(buffer.get(), size), &size, io_ctx); + auto st = remote_file_reader->read_at(empty_start, Slice(buffer.get(), size), &size, io_ctx); + if (st.ok()) { + stats.remote_io_sample_bytes += size; + } + return st; } CloudWarmUpManager& get_warm_up_manager() { @@ -1039,6 +1044,7 @@ Status CachedRemoteFileReader::_read_remaining_blocks_from_cache( RETURN_IF_ERROR(_remote_file_reader->read_at( current_offset, Slice(result.data + (current_offset - offset), read_size), &remote_bytes_read, io_ctx)); + stats.remote_io_sample_bytes += remote_bytes_read; indirect_read_bytes += read_size; source_read_breakdown.remote_bytes += remote_bytes_read; DCHECK(remote_bytes_read == read_size); @@ -1113,6 +1119,7 @@ Status CachedRemoteFileReader::_read_remote_only_on_cache_miss( SCOPED_RAW_TIMER(&stats.remote_read_timer); RETURN_IF_ERROR(_remote_file_reader->read_at(offset, Slice(result.data, bytes_req), &remote_bytes_read, io_ctx)); + stats.remote_io_sample_bytes += remote_bytes_read; *bytes_read = remote_bytes_read; DCHECK_EQ(*bytes_read, bytes_req); source_read_breakdown.remote_bytes += remote_bytes_read; @@ -1180,6 +1187,100 @@ 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) { + 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; + } + // 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; + return Status::OK(); +} + Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_read, const IOContext* io_ctx) { SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().cached_remote_reader_read_at); @@ -1222,6 +1323,13 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t* path().native(), offset, bytes_req, read_at_sw.elapsed_time_milliseconds(), io_ctx->is_warmup); } + if (read_st.ok() && _io_adaptor_eligible(bytes_req, io_ctx) && + stats.remote_io_sample_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_io_sample_bytes), stats.remote_read_timer, + config::file_cache_each_block_size)); + } if (read_st.ok() && !is_dryrun) { // Only successful reads contribute to query profile and file-cache metrics. const auto file_cache_read_type = @@ -1250,6 +1358,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; + } + } if (use_remote_only_on_cache_miss(io_ctx)) { read_st = _read_remote_only_on_cache_miss(offset, result, bytes_req, is_dryrun, bytes_read, stats, source_read_breakdown, io_ctx); @@ -1380,6 +1496,18 @@ void CachedRemoteFileReader::_update_stats(const ReadStatistics& read_stats, statis->lock_wait_timer += read_stats.lock_wait_timer; statis->get_timer += read_stats.get_timer; statis->set_timer += read_stats.set_timer; + statis->io_adaptor_lookup_count += read_stats.io_adaptor_lookup_count; + statis->io_adaptor_lookup_time_ns += read_stats.io_adaptor_lookup_time_ns; + statis->io_adaptor_local_reads += read_stats.io_adaptor_local_reads; + statis->io_adaptor_local_time_ns += read_stats.io_adaptor_local_time_ns; + statis->io_adaptor_bypass_count += read_stats.io_adaptor_bypass_count; + statis->io_adaptor_bypass_bytes += read_stats.io_adaptor_bypass_bytes; + statis->io_adaptor_remote_bytes += read_stats.io_adaptor_remote_bytes; + statis->io_adaptor_remote_time_ns += read_stats.io_adaptor_remote_time_ns; + statis->io_adaptor_local_feedbacks += read_stats.io_adaptor_local_feedbacks; + statis->io_adaptor_remote_feedbacks += read_stats.io_adaptor_remote_feedbacks; + statis->io_adaptor_remote_failures += read_stats.io_adaptor_remote_failures; + statis->io_adaptor_fallback_local += read_stats.io_adaptor_fallback_local; statis->async_cache_write_submitted += read_stats.async_cache_write_submitted; statis->async_cache_write_rejected += read_stats.async_cache_write_rejected; statis->async_cache_write_buffer_alloc_fail += read_stats.async_cache_write_buffer_alloc_fail; diff --git a/be/src/io/cache/cached_remote_file_reader.h b/be/src/io/cache/cached_remote_file_reader.h index 68b112d9ece..72803cb4717 100644 --- a/be/src/io/cache/cached_remote_file_reader.h +++ b/be/src/io/cache/cached_remote_file_reader.h @@ -426,6 +426,12 @@ private: const SourceReadBreakdown& source_read_breakdown, FileCacheStatistics* state, FileCacheReadType read_type) const; + bool _io_adaptor_eligible(size_t bytes, const IOContext* io_ctx) const; + Status _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); + bool _is_doris_table = false; CacheAlignMode _cache_align_mode {CacheAlignMode::ALIGN_TO_BLOCK}; CacheWriteMode _cache_write_mode {CacheWriteMode::DEFAULT}; diff --git a/be/src/io/cache/file_block.cpp b/be/src/io/cache/file_block.cpp index d9d57b08bf9..f221d9eb932 100644 --- a/be/src/io/cache/file_block.cpp +++ b/be/src/io/cache/file_block.cpp @@ -230,6 +230,7 @@ Status FileBlock::finalize() { } Status FileBlock::read(Slice buffer, size_t read_offset) { + TEST_SYNC_POINT("FileBlock::adaptor_benchmark_read"); return _mgr->_storage->read(_key, read_offset, buffer); } diff --git a/be/src/io/cache/file_cache_common.h b/be/src/io/cache/file_cache_common.h index 554d049b21e..514d3173000 100644 --- a/be/src/io/cache/file_cache_common.h +++ b/be/src/io/cache/file_cache_common.h @@ -78,6 +78,7 @@ struct ReadStatistics { int64_t bytes_read_from_peer = 0; int64_t bytes_write_into_file_cache = 0; int64_t remote_read_timer = 0; + int64_t remote_io_sample_bytes = 0; int64_t peer_read_timer = 0; int64_t remote_wait_timer = 0; // wait for other downloader int64_t local_read_timer = 0; @@ -87,6 +88,18 @@ struct ReadStatistics { int64_t lock_wait_timer = 0; int64_t get_timer = 0; int64_t set_timer = 0; + int64_t io_adaptor_lookup_count = 0; + int64_t io_adaptor_lookup_time_ns = 0; + int64_t io_adaptor_local_reads = 0; + int64_t io_adaptor_local_time_ns = 0; + int64_t io_adaptor_bypass_count = 0; + int64_t io_adaptor_bypass_bytes = 0; + int64_t io_adaptor_remote_bytes = 0; + int64_t io_adaptor_remote_time_ns = 0; + int64_t io_adaptor_local_feedbacks = 0; + int64_t io_adaptor_remote_feedbacks = 0; + int64_t io_adaptor_remote_failures = 0; + int64_t io_adaptor_fallback_local = 0; int64_t async_cache_write_submitted = 0; int64_t async_cache_write_rejected = 0; int64_t async_cache_write_buffer_alloc_fail = 0; diff --git a/be/src/io/cache/file_cache_io_adaptor.cpp b/be/src/io/cache/file_cache_io_adaptor.cpp new file mode 100644 index 00000000000..b2b20a4f2a0 --- /dev/null +++ b/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; + } + // The unused path contributes its remembered minimum, not a fabricated real IO sample. + // This is SR's feedback mechanism; feedback must not refresh 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; + } + // Keep the public floating-point factor intact rather than truncating fractional values. + 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(); + } else { + decision.remote_feedback = _remote.feedback(); + } + return decision; +} + +void FileCacheIOAdaptor::record_local(int64_t latency_us) { + std::lock_guard lock(_mutex); + _local.record(latency_us); +} + +void FileCacheIOAdaptor::record_remote(int64_t latency_us) { + std::lock_guard lock(_mutex); + _remote.record(latency_us); +} + +int64_t FileCacheIOAdaptor::latency_per_block_us(size_t bytes, int64_t elapsed_ns, + size_t block_size) { + if (bytes == 0 || block_size == 0 || elapsed_ns < 0) { + return -1; + } + const auto latency_us = static_cast<uint64_t>(elapsed_ns / 1000); + const auto scaled = static_cast<unsigned __int128>(latency_us) * block_size / bytes; + unsigned __int128 estimate = latency_us; + if (bytes > block_size && bytes - block_size > block_size) { + estimate = std::min(estimate, scaled * 2); + } else if (bytes < block_size && bytes < block_size - bytes) { + estimate = std::max(estimate, scaled / 2); + } + return static_cast<int64_t>(std::min( + estimate, static_cast<unsigned __int128>(std::numeric_limits<int64_t>::max()))); +} + +} // namespace doris::io diff --git a/be/src/io/cache/file_cache_io_adaptor.h b/be/src/io/cache/file_cache_io_adaptor.h new file mode 100644 index 00000000000..bb0f6895218 --- /dev/null +++ b/be/src/io/cache/file_cache_io_adaptor.h @@ -0,0 +1,72 @@ +// 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. + +#pragma once + +#include <array> +#include <cstddef> +#include <cstdint> +#include <limits> +#include <mutex> + +namespace doris::io { + +// One BE-wide history matches the shared cache-engine scope of the external-read adaptor. +class FileCacheIOAdaptor { +public: + struct Decision { + bool bypass = false; + bool local_feedback = false; + bool remote_feedback = false; + int64_t local_latency_us = 0; + int64_t remote_latency_us = 0; + bool read_remote() const { return bypass; } + }; + + static FileCacheIOAdaptor& instance(); + Decision select(double skip_read_factor = 1.0); + void record_local(int64_t latency_us); + void record_remote(int64_t latency_us); + + // Estimate service time for one native cache block using the same empirical ratio as SR. + // Normalize both sides because Doris can read a partial local block without a block buffer. + static int64_t latency_per_block_us(size_t bytes, int64_t elapsed_ns, size_t block_size); + +private: + static constexpr size_t WINDOW_SIZE = 50; + static constexpr int64_t MIN_LOCAL_LATENCY_US = 5000; + static constexpr int64_t UPPER_MIN_LATENCY_US = 100'000'000; + static constexpr uint64_t MINIMUM_LIFE_LIMIT = 500'000; + struct Window { + std::array<int64_t, WINDOW_SIZE> samples {}; + size_t count = 0; + size_t next = 0; + // Wide accumulation preserves the average even for unusually long, valid IO durations. + __int128 sum = 0; + int64_t minimum = std::numeric_limits<int64_t>::max(); + uint64_t minimum_life = 0; + void record(int64_t latency_us); + void append(int64_t latency_us); + bool feedback(); + int64_t average() const; + }; + std::mutex _mutex; + Window _local; + Window _remote; +}; + +} // namespace doris::io diff --git a/be/src/io/io_common.h b/be/src/io/io_common.h index 0109a345af3..b43a55426ef 100644 --- a/be/src/io/io_common.h +++ b/be/src/io/io_common.h @@ -73,6 +73,18 @@ struct FileCacheStatistics { int64_t lock_wait_timer = 0; int64_t get_timer = 0; int64_t set_timer = 0; + int64_t io_adaptor_lookup_count = 0; + int64_t io_adaptor_lookup_time_ns = 0; + int64_t io_adaptor_local_reads = 0; + int64_t io_adaptor_local_time_ns = 0; + int64_t io_adaptor_bypass_count = 0; + int64_t io_adaptor_bypass_bytes = 0; + int64_t io_adaptor_remote_bytes = 0; + int64_t io_adaptor_remote_time_ns = 0; + int64_t io_adaptor_local_feedbacks = 0; + int64_t io_adaptor_remote_feedbacks = 0; + int64_t io_adaptor_remote_failures = 0; + int64_t io_adaptor_fallback_local = 0; int64_t async_cache_write_submitted = 0; int64_t async_cache_write_rejected = 0; int64_t async_cache_write_buffer_alloc_fail = 0; @@ -145,6 +157,18 @@ struct FileCacheStatistics { lock_wait_timer += other.lock_wait_timer; get_timer += other.get_timer; set_timer += other.set_timer; + io_adaptor_lookup_count += other.io_adaptor_lookup_count; + io_adaptor_lookup_time_ns += other.io_adaptor_lookup_time_ns; + io_adaptor_local_reads += other.io_adaptor_local_reads; + io_adaptor_local_time_ns += other.io_adaptor_local_time_ns; + io_adaptor_bypass_count += other.io_adaptor_bypass_count; + io_adaptor_bypass_bytes += other.io_adaptor_bypass_bytes; + io_adaptor_remote_bytes += other.io_adaptor_remote_bytes; + io_adaptor_remote_time_ns += other.io_adaptor_remote_time_ns; + io_adaptor_local_feedbacks += other.io_adaptor_local_feedbacks; + io_adaptor_remote_feedbacks += other.io_adaptor_remote_feedbacks; + io_adaptor_remote_failures += other.io_adaptor_remote_failures; + io_adaptor_fallback_local += other.io_adaptor_fallback_local; async_cache_write_submitted += other.async_cache_write_submitted; async_cache_write_rejected += other.async_cache_write_rejected; async_cache_write_buffer_alloc_fail += other.async_cache_write_buffer_alloc_fail; @@ -213,6 +237,8 @@ struct IOContext { bool is_disposable = false; bool is_index_data = false; bool read_file_cache = true; + // Only explicitly enabled query IO may use adaptation; warmup and background IO default off. + bool enable_file_cache_io_adaptor = false; // TODO(lightman): use following member variables to control file cache bool is_persistent = false; // stop reader when reading, used in some interrupted operations diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index ef7c6ab19e2..9a777a3d866 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -1143,6 +1143,32 @@ TEST(FileScannerV2Test, FileScanIoContextDoesNotMarkLoadAsQueryReader) { EXPECT_EQ(io_ctx->remote_scan_cache_write_limiter, nullptr); } +// Both scanner generations use this helper, so query options must survive FE/BE version skew. +TEST(FileScannerV2Test, FileScanIoContextUsesExplicitQueryAdaptorOption) { + MockRuntimeState state; + state._query_id = make_query_id(); + state._query_ctx = nullptr; + TQueryOptions options; + options.__set_query_type(TQueryType::SELECT); + state.set_query_options(options); + EXPECT_FALSE(create_file_scan_io_context(&state)->enable_file_cache_io_adaptor); + + options.__set_enable_file_cache_io_adaptor(true); + state.set_query_options(options); + auto enabled = create_file_scan_io_context(&state); + EXPECT_TRUE(enabled->enable_file_cache_io_adaptor); + + options.__set_enable_file_cache_io_adaptor(false); + state.set_query_options(options); + EXPECT_FALSE(create_file_scan_io_context(&state)->enable_file_cache_io_adaptor); + EXPECT_TRUE(enabled->enable_file_cache_io_adaptor); + + options.__set_query_type(TQueryType::LOAD); + options.__set_enable_file_cache_io_adaptor(true); + state.set_query_options(options); + EXPECT_FALSE(create_file_scan_io_context(&state)->enable_file_cache_io_adaptor); +} + // Scenario: FileScannerV2 converts only the file formats implemented by format_v2 readers and // rejects everything else before TableReader::init sees an unsupported FileFormat. TEST(FileScannerV2Test, FileFormatConversionMatrix) { diff --git a/be/test/io/cache/block_file_cache_profile_reporter_test.cpp b/be/test/io/cache/block_file_cache_profile_reporter_test.cpp index 4836bcb2638..9a65e3b805b 100644 --- a/be/test/io/cache/block_file_cache_profile_reporter_test.cpp +++ b/be/test/io/cache/block_file_cache_profile_reporter_test.cpp @@ -54,6 +54,18 @@ io::FileCacheStatistics make_file_cache_stats(int64_t multiplier) { stats.inverted_index_io_timer = multiplier * 28; stats.remote_only_on_miss_triggered = multiplier * 29; stats.remote_only_on_miss_threshold_bytes = multiplier * 30; + stats.io_adaptor_lookup_count = multiplier * 60; + stats.io_adaptor_lookup_time_ns = multiplier * 61; + stats.io_adaptor_local_reads = multiplier * 62; + stats.io_adaptor_local_time_ns = multiplier * 63; + stats.io_adaptor_bypass_count = multiplier * 64; + stats.io_adaptor_bypass_bytes = multiplier * 65; + stats.io_adaptor_remote_bytes = multiplier * 66; + stats.io_adaptor_remote_time_ns = multiplier * 67; + stats.io_adaptor_local_feedbacks = multiplier * 68; + stats.io_adaptor_remote_feedbacks = multiplier * 69; + stats.io_adaptor_remote_failures = multiplier * 70; + stats.io_adaptor_fallback_local = multiplier * 71; stats.async_cache_write_submitted = multiplier * 45; stats.async_cache_write_rejected = multiplier * 46; stats.async_cache_write_buffer_alloc_fail = multiplier * 47; @@ -70,6 +82,18 @@ io::FileCacheStatistics make_file_cache_stats(int64_t multiplier) { void expect_file_cache_stats_eq(const io::FileCacheStatistics& actual, const io::FileCacheStatistics& expected) { + EXPECT_EQ(actual.io_adaptor_lookup_count, expected.io_adaptor_lookup_count); + EXPECT_EQ(actual.io_adaptor_lookup_time_ns, expected.io_adaptor_lookup_time_ns); + EXPECT_EQ(actual.io_adaptor_local_reads, expected.io_adaptor_local_reads); + EXPECT_EQ(actual.io_adaptor_local_time_ns, expected.io_adaptor_local_time_ns); + EXPECT_EQ(actual.io_adaptor_bypass_count, expected.io_adaptor_bypass_count); + EXPECT_EQ(actual.io_adaptor_bypass_bytes, expected.io_adaptor_bypass_bytes); + EXPECT_EQ(actual.io_adaptor_remote_bytes, expected.io_adaptor_remote_bytes); + EXPECT_EQ(actual.io_adaptor_remote_time_ns, expected.io_adaptor_remote_time_ns); + EXPECT_EQ(actual.io_adaptor_local_feedbacks, expected.io_adaptor_local_feedbacks); + EXPECT_EQ(actual.io_adaptor_remote_feedbacks, expected.io_adaptor_remote_feedbacks); + EXPECT_EQ(actual.io_adaptor_remote_failures, expected.io_adaptor_remote_failures); + EXPECT_EQ(actual.io_adaptor_fallback_local, expected.io_adaptor_fallback_local); EXPECT_EQ(actual.num_local_io_total, expected.num_local_io_total); EXPECT_EQ(actual.num_remote_io_total, expected.num_remote_io_total); EXPECT_EQ(actual.num_peer_io_total, expected.num_peer_io_total); @@ -165,6 +189,30 @@ TEST(FileCacheProfileReporterTest, ReporterAggregatesDeltaReportsToExactFinalTot io::diff_file_cache_statistics(after_second_report, after_first_report); reporter.update(&second_delta); + EXPECT_EQ(profile->get_counter("IOAdaptorLookupCount")->value(), + after_second_report.io_adaptor_lookup_count); + EXPECT_EQ(profile->get_counter("IOAdaptorLookupTime")->value(), + after_second_report.io_adaptor_lookup_time_ns); + EXPECT_EQ(profile->get_counter("IOAdaptorLocalReadCount")->value(), + after_second_report.io_adaptor_local_reads); + EXPECT_EQ(profile->get_counter("IOAdaptorLocalReadTime")->value(), + after_second_report.io_adaptor_local_time_ns); + EXPECT_EQ(profile->get_counter("IOAdaptorBypassCount")->value(), + after_second_report.io_adaptor_bypass_count); + EXPECT_EQ(profile->get_counter("IOAdaptorBypassBytes")->value(), + after_second_report.io_adaptor_bypass_bytes); + EXPECT_EQ(profile->get_counter("IOAdaptorRemoteReadBytes")->value(), + after_second_report.io_adaptor_remote_bytes); + EXPECT_EQ(profile->get_counter("IOAdaptorRemoteReadTime")->value(), + after_second_report.io_adaptor_remote_time_ns); + EXPECT_EQ(profile->get_counter("IOAdaptorLocalFeedbackCount")->value(), + after_second_report.io_adaptor_local_feedbacks); + EXPECT_EQ(profile->get_counter("IOAdaptorRemoteFeedbackCount")->value(), + after_second_report.io_adaptor_remote_feedbacks); + EXPECT_EQ(profile->get_counter("IOAdaptorRemoteFailureCount")->value(), + after_second_report.io_adaptor_remote_failures); + EXPECT_EQ(profile->get_counter("IOAdaptorFallbackLocalCount")->value(), + after_second_report.io_adaptor_fallback_local); EXPECT_EQ(profile->get_counter("BytesScannedFromCache")->value(), after_second_report.bytes_read_from_local); EXPECT_EQ(profile->get_counter("BytesScannedFromRemote")->value(), diff --git a/be/test/io/cache/cached_remote_file_reader_test.cpp b/be/test/io/cache/cached_remote_file_reader_test.cpp index 6fe9bac5d60..418f6d60e8e 100644 --- a/be/test/io/cache/cached_remote_file_reader_test.cpp +++ b/be/test/io/cache/cached_remote_file_reader_test.cpp @@ -15,12 +15,17 @@ // specific language governing permissions and limitations // under the License. +#include <algorithm> +#include <chrono> #include <cstring> #include <future> +#include <iostream> +#include <thread> #include "block_file_cache_test_common.h" #include "cloud/config.h" #include "io/cache/async_cache_write_manager_metrics.h" +#include "io/cache/file_cache_io_adaptor.h" #include "io/cache/inflight_write_buffer_index.h" #include "util/time.h" @@ -57,6 +62,10 @@ public: bool closed() const override { return _delegate->closed(); } int64_t mtime() const override { return _delegate->mtime(); } + int64_t delay_us = 0; + bool fail_reads = false; + bool cancel_reads = false; + size_t short_read_size = 0; size_t read_count() const { return _read_count; } size_t last_offset() const { return _last_offset; } size_t last_size() const { return _last_size; } @@ -69,6 +78,18 @@ protected: _last_offset = offset; _last_size = result.size; _last_bypass_peer_read = io_ctx != nullptr && io_ctx->bypass_peer_read; + if (delay_us > 0) { + std::this_thread::sleep_for(std::chrono::microseconds(delay_us)); + } + if (cancel_reads) { + return Status::Cancelled("Injected remote cancellation"); + } + if (fail_reads) { + return Status::IOError("Injected remote read failure"); + } + if (short_read_size != 0) { + result = Slice(result.data, std::min(result.size, short_read_size)); + } return _delegate->read_at(offset, result, bytes_read, io_ctx); } @@ -1661,4 +1682,467 @@ TEST_F(BlockFileCacheTest, cache_write_mode_is_resolved_for_each_read_context) { EXPECT_EQ(sync_reader->_resolve_cache_write_mode(&context), CacheWriteMode::ASYNC_WRITE); } +// All-hit adaptation is shared by sync/async readers. Cold and mixed requests retain their +// original downloader/inflight/writeback semantics rather than pretending a cache hit is EMPTY. +class AdaptiveCachedRemoteFileReaderTest : public AsyncCachedRemoteFileReaderTest { +protected: + void SetUp() override { + AsyncCachedRemoteFileReaderTest::SetUp(); + reset_adaptor(); + } + void reset_adaptor() { + auto& adaptor = FileCacheIOAdaptor::instance(); + std::lock_guard lock(adaptor._mutex); + adaptor._local = {}; + adaptor._remote = {}; + } + std::shared_ptr<CachedRemoteFileReader> external_reader(FileReaderSPtr remote) { + FileReaderOptions options; + options.cache_type = FileCachePolicy::FILE_BLOCK_CACHE; + return std::make_shared<CachedRemoteFileReader>(std::move(remote), options); + } + void seed_slow_disk() { + auto& adaptor = FileCacheIOAdaptor::instance(); + for (int i = 0; i < 50; ++i) { + adaptor.record_local(20000); + adaptor.record_remote(1000); + } + } + void warm_adaptor_with_real_reads(const std::shared_ptr<CachedRemoteFileReader>& reader, + const std::shared_ptr<CountingFileReader>& remote, + size_t bytes) { + // Each benchmark group starts with measured service times in the shared controller. + // Calibration is excluded from the timed steady-state phase in both configurations. + reset_adaptor(); + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + context.enable_file_cache_io_adaptor = true; + std::string buffer(bytes, '\0'); + size_t count = 0; + for (int sample = 0; sample < 50; ++sample) { + const auto start = MonotonicNanos(); + ASSERT_TRUE(remote->read_at(0, Slice(buffer), &count, &context).ok()); + ASSERT_EQ(count, bytes); + FileCacheIOAdaptor::instance().record_remote(FileCacheIOAdaptor::latency_per_block_us( + bytes, MonotonicNanos() - start, config::file_cache_each_block_size)); + } + for (int sample = 0; sample < 50; ++sample) { + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + ASSERT_EQ(count, bytes); + } + } +}; + +TEST_F(AdaptiveCachedRemoteFileReaderTest, full_hit_bypass_preserves_cache_in_both_write_modes) { + create_cache("adaptive_full_hit"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(17, Slice(expected), &count, &context).ok()); + ASSERT_EQ(count, expected.size()); + seed_slow_disk(); + context.enable_file_cache_io_adaptor = true; + const auto cached_blocks = cache()->get_blocks_by_key(reader->_cache_hash).size(); + for (auto mode : {CacheWriteMode::SYNC_WRITE, CacheWriteMode::ASYNC_WRITE}) { + FileCacheStatistics stats; + context.file_cache_stats = &stats; + context.cache_write_mode_override = mode; + const auto previous_reads = remote->read_count(); + std::string actual(expected.size(), '\0'); + ASSERT_TRUE(reader->read_at(17, Slice(actual), &count, &context).ok()); + EXPECT_EQ(actual, expected); + EXPECT_EQ(remote->read_count(), previous_reads + 1); + EXPECT_EQ(remote->last_offset(), 17); + EXPECT_EQ(remote->last_size(), actual.size()); + EXPECT_EQ(stats.io_adaptor_bypass_count, 1); + EXPECT_EQ(stats.io_adaptor_bypass_bytes, actual.size()); + EXPECT_EQ(stats.io_adaptor_remote_bytes, actual.size()); + EXPECT_EQ(stats.bytes_read_from_remote, actual.size()); + EXPECT_EQ(stats.bytes_write_into_cache, 0); + EXPECT_EQ(stats.async_cache_write_submitted, 0); + EXPECT_EQ(cache()->get_blocks_by_key(reader->_cache_hash).size(), cached_blocks); + } +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, cross_block_and_eof_ranges_match_on_both_routes) { + create_cache("adaptive_read_ranges"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + for (size_t offset : {size_t(1_mb - 17), reader->size() - 31}) { + reset_adaptor(); + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + std::string expected(64 * 1024, '\0'); + size_t count = 0; + ASSERT_TRUE(reader->read_at(offset, Slice(expected), &count, &context).ok()); + expected.resize(count); + context.enable_file_cache_io_adaptor = true; + for (bool bypass : {false, true}) { + if (bypass) { + seed_slow_disk(); + } + const auto previous_reads = remote->read_count(); + FileCacheStatistics stats; + context.file_cache_stats = &stats; + std::string actual(64 * 1024, '\xff'); + ASSERT_TRUE(reader->read_at(offset, Slice(actual), &count, &context).ok()); + EXPECT_EQ(count, expected.size()); + EXPECT_EQ(actual.substr(0, count), expected); + EXPECT_EQ(actual.substr(count), std::string(actual.size() - count, '\xff')); + EXPECT_EQ(remote->read_count(), previous_reads + size_t(bypass)); + EXPECT_EQ(stats.io_adaptor_bypass_count, int64_t(bypass)); + EXPECT_EQ(stats.io_adaptor_local_reads, int64_t(!bypass)); + EXPECT_EQ(stats.bytes_write_into_cache, 0); + } + } +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, disabled_and_metadata_reads_preserve_local_path) { + create_cache("adaptive_disabled"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string buffer(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + seed_slow_disk(); + const auto previous_reads = remote->read_count(); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), previous_reads); + context.enable_file_cache_io_adaptor = true; + context.is_index_data = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), previous_reads); + context.is_index_data = false; + context.is_warmup = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + context.is_warmup = false; + context.is_dryrun = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + context.is_dryrun = false; + context.should_stop = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + context.should_stop = false; + context.read_file_cache = false; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + context.read_file_cache = true; + context.is_inverted_index = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + context.is_inverted_index = false; + EXPECT_FALSE(reader->_io_adaptor_eligible(0, &context)); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, query_override_does_not_change_other_readers) { + create_cache("adaptive_query_override"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string buffer(64 * 1024, '\0'); + size_t count = 0; + IOContext enabled; + enabled.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &enabled).ok()); + seed_slow_disk(); + enabled.enable_file_cache_io_adaptor = true; + IOContext disabled; + disabled.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + const auto before = remote->read_count(); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &enabled).ok()); + EXPECT_EQ(remote->read_count(), before + 1); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &disabled).ok()); + EXPECT_EQ(remote->read_count(), before + 1); + // An internal-table reader cannot opt into the external-query policy accidentally. + FileReaderOptions options; + options.cache_type = FileCachePolicy::FILE_BLOCK_CACHE; + options.is_doris_table = true; + auto internal = std::make_shared<CachedRemoteFileReader>(remote, options); + EXPECT_FALSE(internal->_io_adaptor_eligible(buffer.size(), &enabled)); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, real_remote_misses_train_only_enabled_queries) { + create_cache("adaptive_remote_history"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string buffer(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + auto& adaptor = FileCacheIOAdaptor::instance(); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(adaptor._remote.count, 0); + context.enable_file_cache_io_adaptor = true; + size_t offset = 0; + for (auto mode : {CacheWriteMode::SYNC_WRITE, CacheWriteMode::ASYNC_WRITE}) { + offset += 1_mb; + reset_adaptor(); + context.cache_write_mode_override = mode; + ASSERT_TRUE(reader->read_at(offset, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(adaptor._remote.count, 1); + EXPECT_EQ(adaptor._local.count, 0); + wait_for_async_writes(); + const auto reads = remote->read_count(); + ASSERT_TRUE(reader->read_at(offset, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), reads); + EXPECT_EQ(adaptor._remote.count, 1); + EXPECT_EQ(adaptor._local.count, 1); + } +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, remote_failure_and_short_read_fall_back_to_valid_cache) { + create_cache("adaptive_remote_failure"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(7, Slice(expected), &count, &context).ok()); + seed_slow_disk(); + context.enable_file_cache_io_adaptor = true; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + remote->short_read_size = 7; + std::string actual(expected.size(), '\xff'); + ASSERT_TRUE(reader->read_at(7, Slice(actual), &count, &context).ok()); + EXPECT_EQ(actual, expected); + EXPECT_EQ(count, expected.size()); + EXPECT_EQ(stats.io_adaptor_fallback_local, 1); + EXPECT_EQ(stats.io_adaptor_remote_bytes, 7); + EXPECT_EQ(stats.bytes_read_from_local, expected.size()); + EXPECT_EQ(stats.bytes_write_into_cache, 0); + EXPECT_EQ(stats.io_adaptor_remote_failures, 1); + // A failed speculative read never damages the cached range, including consecutive errors. + remote->short_read_size = 0; + remote->fail_reads = true; + const auto previous_reads = remote->read_count(); + ASSERT_TRUE(reader->read_at(7, Slice(actual), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), previous_reads + 1); + EXPECT_EQ(stats.io_adaptor_fallback_local, 2); + EXPECT_EQ(stats.io_adaptor_remote_failures, 2); + EXPECT_EQ(actual, expected); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, cancellation_does_not_fall_back_or_rewrite_cache) { + create_cache("adaptive_cancelled"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string buffer(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + seed_slow_disk(); + context.enable_file_cache_io_adaptor = true; + remote->cancel_reads = true; + const auto previous_reads = remote->read_count(); + auto status = reader->read_at(0, Slice(buffer), &count, &context); + EXPECT_TRUE(status.is<ErrorCode::CANCELLED>()); + EXPECT_EQ(remote->read_count(), previous_reads + 1); + // Cancellation is propagated, while a subsequent query can still use the intact cache. + remote->cancel_reads = false; + context.enable_file_cache_io_adaptor = false; + const auto reads_after_cancel = remote->read_count(); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), reads_after_cancel); + EXPECT_EQ(count, buffer.size()); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, mixed_hit_miss_and_eof_keep_existing_async_semantics) { + create_cache("adaptive_mixed_range"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string first(1_mb, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(first), &count, &context).ok()); + seed_slow_disk(); + context.enable_file_cache_io_adaptor = true; + context.cache_write_mode_override = CacheWriteMode::ASYNC_WRITE; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + std::string actual(2_mb, '\0'); + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + EXPECT_EQ(count, actual.size()); + EXPECT_EQ(actual.substr(0, first.size()), first); + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); + EXPECT_GT(stats.async_cache_write_submitted, 0); + wait_for_async_writes(); + ASSERT_TRUE(reader->read_at(reader->size(), Slice(actual), &count, &context).ok()); + EXPECT_EQ(count, 0); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, missing_cache_file_retries_the_complete_range) { + create_cache("adaptive_missing_file"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(expected), &count, &context).ok()); + auto blocks = cache()->get_blocks_by_key(reader->_cache_hash); + ASSERT_FALSE(blocks.empty()); + std::error_code error; + ASSERT_TRUE(fs::remove(blocks.begin()->second->get_cache_file(), error)); + ASSERT_FALSE(error); + blocks.clear(); + context.enable_file_cache_io_adaptor = true; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + const auto previous_reads = remote->read_count(); + std::string actual(expected.size(), '\xff'); + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + EXPECT_EQ(actual, expected); + EXPECT_EQ(count, actual.size()); + EXPECT_EQ(remote->read_count(), previous_reads + 1); + EXPECT_EQ(stats.io_adaptor_local_reads, 1); + EXPECT_GT(stats.io_adaptor_local_time_ns, 0); + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, inflight_payload_still_avoids_remote_reads) { + create_cache("adaptive_inflight"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + auto* index = cache()->inflight_write_buffer_index(); + AsyncCacheWriteBufferPtr payload; + ASSERT_TRUE(cache()->async_write_manager()->allocate_tracked_buffer(1_mb, &payload).ok()); + const std::string expected(1_mb, '0'); + memcpy(payload->data(), expected.data(), expected.size()); + auto entry = std::make_shared<InflightWriteBufferEntry>(payload, 0, expected.size(), + MonotonicMicros()); + ASSERT_EQ(index->insert_if_absent(reader->_cache_hash, 0, entry), nullptr); + Defer remove_entry { + [&] { static_cast<void>(index->remove_if(reader->_cache_hash, 0, entry)); }}; + seed_slow_disk(); + FileCacheStatistics stats; + IOContext context; + context.enable_file_cache_io_adaptor = true; + context.file_cache_stats = &stats; + context.cache_write_mode_override = CacheWriteMode::ASYNC_WRITE; + std::string actual(expected.size(), '\0'); + size_t count = 0; + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + EXPECT_EQ(actual, expected); + EXPECT_EQ(count, actual.size()); + EXPECT_EQ(remote->read_count(), 0); + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); + EXPECT_GT(stats.inflight_write_buffer_index_hit, 0); + EXPECT_EQ(stats.async_cache_write_submitted, 0); +} + +// Opt-in end-to-end cache-path benchmark. Injected service times model a congested cache disk; +// the measured wall times are a controlled local experiment, not a production query forecast. +TEST_F(AdaptiveCachedRemoteFileReaderTest, DISABLED_latency_controlled_benchmark) { + create_cache("adaptive_latency_benchmark"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(expected), &count, &context).ok()); + remote->delay_us = 1000; + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard guard; + sync_point->set_call_back( + "FileBlock::adaptor_benchmark_read", + [](auto&&) { std::this_thread::sleep_for(std::chrono::microseconds(8000)); }, &guard); + sync_point->enable_processing(); + Defer cleanup {[&] { sync_point->disable_processing(); }}; + std::vector<int64_t> off_times; + std::vector<int64_t> on_times; + for (int round = 0; round < 5; ++round) { + for (int phase = 0; phase < 2; ++phase) { + const bool enabled = (round + phase) % 2 == 1; + ASSERT_NO_FATAL_FAILURE(warm_adaptor_with_real_reads(reader, remote, expected.size())); + context.enable_file_cache_io_adaptor = enabled; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + std::string actual(expected.size(), '\0'); + const auto start = MonotonicNanos(); + for (int request = 0; request < 200; ++request) { + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + ASSERT_EQ(actual, expected); + ASSERT_EQ(count, expected.size()); + } + const auto elapsed = MonotonicNanos() - start; + (enabled ? on_times : off_times).push_back(elapsed); + std::cout << "IO_ADAPTOR_BENCH round=" << round << " enabled=" << enabled + << " requests=200 request_bytes=" << expected.size() + << " local_service_us=8000 remote_service_us=1000 wall_ns=" << elapsed + << " bypass_count=" << stats.io_adaptor_bypass_count + << " bypass_bytes=" << stats.io_adaptor_bypass_bytes + << " local_feedback_count=" << stats.io_adaptor_local_feedbacks << std::endl; + EXPECT_EQ(stats.bytes_write_into_cache, 0); + if (enabled) { + EXPECT_GT(stats.io_adaptor_bypass_count, 180); + EXPECT_GT(stats.io_adaptor_local_feedbacks, 0); + } else { + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); + } + } + } + std::sort(off_times.begin(), off_times.end()); + std::sort(on_times.begin(), on_times.end()); + std::cout << "IO_ADAPTOR_BENCH_MEDIAN off_ns=" << off_times[2] << " on_ns=" << on_times[2] + << std::endl; +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, DISABLED_fast_local_latency_controlled_benchmark) { + create_cache("adaptive_fast_local_benchmark"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(expected), &count, &context).ok()); + remote->delay_us = 8000; + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard guard; + sync_point->set_call_back( + "FileBlock::adaptor_benchmark_read", + [](auto&&) { std::this_thread::sleep_for(std::chrono::microseconds(1000)); }, &guard); + sync_point->enable_processing(); + Defer cleanup {[&] { sync_point->disable_processing(); }}; + std::vector<int64_t> off_times; + std::vector<int64_t> on_times; + for (int round = 0; round < 5; ++round) { + for (int phase = 0; phase < 2; ++phase) { + const bool enabled = (round + phase) % 2 == 1; + ASSERT_NO_FATAL_FAILURE(warm_adaptor_with_real_reads(reader, remote, expected.size())); + context.enable_file_cache_io_adaptor = enabled; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + std::string actual(expected.size(), '\0'); + const auto previous_reads = remote->read_count(); + const auto start = MonotonicNanos(); + for (int request = 0; request < 200; ++request) { + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + ASSERT_EQ(actual, expected); + ASSERT_EQ(count, expected.size()); + } + const auto elapsed = MonotonicNanos() - start; + (enabled ? on_times : off_times).push_back(elapsed); + std::cout << "IO_ADAPTOR_FAST_LOCAL_BENCH round=" << round << " enabled=" << enabled + << " requests=200 request_bytes=" << expected.size() + << " local_service_us=1000 remote_service_us=8000 wall_ns=" << elapsed + << " bypass_count=" << stats.io_adaptor_bypass_count << std::endl; + EXPECT_EQ(remote->read_count(), previous_reads); + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); + EXPECT_EQ(stats.bytes_write_into_cache, 0); + } + } + std::sort(off_times.begin(), off_times.end()); + std::sort(on_times.begin(), on_times.end()); + std::cout << "IO_ADAPTOR_FAST_LOCAL_BENCH_MEDIAN off_ns=" << off_times[2] + << " on_ns=" << on_times[2] << std::endl; +} + } // namespace doris::io diff --git a/be/test/io/cache/file_cache_io_adaptor_test.cpp b/be/test/io/cache/file_cache_io_adaptor_test.cpp new file mode 100644 index 00000000000..9fefe85dc4b --- /dev/null +++ b/be/test/io/cache/file_cache_io_adaptor_test.cpp @@ -0,0 +1,198 @@ +// 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 <gtest/gtest.h> + +#include <thread> +#include <vector> + +namespace doris::io { + +class FileCacheIOAdaptorTest : public testing::Test { +protected: + void seed(size_t count, int64_t local_us = 20000, int64_t remote_us = 1000) { + for (size_t i = 0; i < count; ++i) { + adaptor.record_local(local_us); + adaptor.record_remote(remote_us); + } + } + FileCacheIOAdaptor adaptor; +}; + +TEST_F(FileCacheIOAdaptorTest, RequiresFiftyRealSamplesOnBothPaths) { + seed(49); + EXPECT_FALSE(adaptor.select().read_remote()); + adaptor.record_local(20000); + EXPECT_FALSE(adaptor.select().read_remote()); + adaptor.record_remote(1000); + EXPECT_TRUE(adaptor.select().read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, MissingRemoteHistoryDoesNotGenerateProbes) { + for (int i = 0; i < 100; ++i) { + adaptor.record_local(20000); + const auto decision = adaptor.select(); + EXPECT_FALSE(decision.read_remote()); + EXPECT_FALSE(decision.local_feedback); + EXPECT_FALSE(decision.remote_feedback); + } +} + +TEST_F(FileCacheIOAdaptorTest, FiveMillisecondFloorAndExactBoundary) { + seed(50, 4999); + EXPECT_FALSE(adaptor.select().read_remote()); + seed(50, 5000); + EXPECT_TRUE(adaptor.select().read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, EqualityPrefersLocalAndFractionalFactorIsPreserved) { + seed(50, 8000, 8000); + EXPECT_FALSE(adaptor.select().read_remote()); + seed(50, 8000, 10000); + EXPECT_FALSE(adaptor.select().read_remote()); + EXPECT_TRUE(adaptor.select(1.5).read_remote()); + EXPECT_FALSE(adaptor.select(0.5).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, SlidingWindowReplacesOldSamples) { + seed(50); + EXPECT_TRUE(adaptor.select().read_remote()); + seed(50, 100, 1000); + EXPECT_FALSE(adaptor.select().read_remote()); + EXPECT_EQ(adaptor.select().local_latency_us, 100); +} + +TEST_F(FileCacheIOAdaptorTest, RememberedLocalMinimumEventuallyReconsidersLocal) { + adaptor.record_local(100); + seed(50, 20000, 1000); + EXPECT_TRUE(adaptor.select().local_feedback); + for (int i = 0; i < 49; ++i) { + adaptor.select(); + } + const auto decision = adaptor.select(); + EXPECT_LT(decision.local_latency_us, 5000); + EXPECT_FALSE(decision.read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, RememberedRemoteMinimumReconsidersRemote) { + adaptor.record_remote(1000); + seed(50, 8000, 20000); + EXPECT_TRUE(adaptor.select().remote_feedback); + for (int i = 0; i < 49; ++i) { + adaptor.select(); + } + EXPECT_TRUE(adaptor.select().read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, FeedbackDoesNotAgeOrReplaceObservedMinimum) { + FileCacheIOAdaptor::Window window; + window.record(100); + window.record(200); + const auto age = window.minimum_life; + for (int i = 0; i < 100; ++i) { + EXPECT_TRUE(window.feedback()); + } + EXPECT_EQ(window.minimum, 100); + EXPECT_EQ(window.minimum_life, age); + EXPECT_EQ(window.average(), 100); +} + +TEST_F(FileCacheIOAdaptorTest, RealSamplesExpireRememberedMinimumAfterLifeLimit) { + FileCacheIOAdaptor::Window window; + window.record(100); + for (uint64_t i = 0; i <= FileCacheIOAdaptor::MINIMUM_LIFE_LIMIT; ++i) { + window.record(200); + } + EXPECT_EQ(window.minimum, 100); + window.record(300); + EXPECT_EQ(window.minimum, 300); + EXPECT_EQ(window.minimum_life, 0); +} + +TEST_F(FileCacheIOAdaptorTest, FeedbackRejectsUninitializedOrExcessiveMinimum) { + FileCacheIOAdaptor::Window window; + EXPECT_FALSE(window.feedback()); + window.record(100000001); + EXPECT_FALSE(window.feedback()); + window.record(100000000); + EXPECT_TRUE(window.feedback()); +} + +TEST_F(FileCacheIOAdaptorTest, NegativeObservationsAreIgnoredAndZeroIsValid) { + adaptor.record_local(-1); + adaptor.record_remote(-1); + EXPECT_EQ(adaptor._local.count, 0); + EXPECT_EQ(adaptor._remote.count, 0); + seed(50, 0, 0); + EXPECT_EQ(adaptor._local.count, 50); + EXPECT_EQ(adaptor._local.minimum, 0); + EXPECT_FALSE(adaptor.select().read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, VeryLargeLatenciesDoNotOverflowWindow) { + const auto maximum = std::numeric_limits<int64_t>::max(); + seed(100, maximum, maximum); + const auto decision = adaptor.select(); + EXPECT_EQ(decision.local_latency_us, maximum); + EXPECT_EQ(decision.remote_latency_us, maximum); + EXPECT_FALSE(decision.read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, NormalizationUsesNativeBlockAndEmpiricalRatio) { + constexpr size_t block = 1 << 20; + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(block, 8000000, block), 8000); + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(block * 2, 8000000, block), 8000); + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(block * 4, 8000000, block), 4000); + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(block / 2, 8000000, block), 8000); + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(block / 4, 8000000, block), 16000); + // Match SR's integer division order before multiplying by the empirical ratio. + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(block * 3, 1000000, block), 666); +} + +TEST_F(FileCacheIOAdaptorTest, NormalizationHandlesInvalidInputsAndSaturates) { + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(0, 1000, 1), -1); + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(1, 1000, 0), -1); + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(1, -1, 1), -1); + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(1, 999, 1), 0); + EXPECT_EQ(FileCacheIOAdaptor::latency_per_block_us(1, std::numeric_limits<int64_t>::max(), + std::numeric_limits<size_t>::max()), + std::numeric_limits<int64_t>::max()); +} + +TEST_F(FileCacheIOAdaptorTest, ConcurrentQueriesShareBoundedHistory) { + std::vector<std::thread> threads; + for (int thread = 0; thread < 8; ++thread) { + threads.emplace_back([&] { + for (int i = 0; i < 10000; ++i) { + adaptor.record_local(20000); + adaptor.record_remote(1000); + adaptor.select(); + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + EXPECT_EQ(adaptor._local.count, 50); + EXPECT_EQ(adaptor._remote.count, 50); + EXPECT_EQ(adaptor.select().local_latency_us, 20000); + EXPECT_TRUE(adaptor.select().read_remote()); +} + +} // namespace doris::io diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index f6eaf55e5f1..843d752416d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -531,6 +531,7 @@ public class SessionVariable implements Serializable, Writable { public static final String ENABLE_SNAPSHOT_POINT_QUERY = "enable_snapshot_point_query"; public static final String ENABLE_FILE_CACHE = "enable_file_cache"; + public static final String ENABLE_FILE_CACHE_IO_ADAPTOR = "enable_file_cache_io_adaptor"; public static final String DISABLE_FILE_CACHE = "disable_file_cache"; @@ -2404,6 +2405,11 @@ public class SessionVariable implements Serializable, Writable { + "The cache is not used when BE config enable_file_cache=false."}) public boolean enableFileCache = true; + @VariableMgr.VarAttr(name = ENABLE_FILE_CACHE_IO_ADAPTOR, needForward = true, description = { + "允许外表查询在本地缓存读取较慢时自适应读取远端数据。", + "Allow external queries to adaptively read remote data when local file-cache reads are slower."}) + public boolean enableFileCacheIoAdaptor = true; + // Specify base path for file cache, or chose a random path. @VariableMgr.VarAttr(name = FILE_CACHE_BASE_PATH, needForward = true, description = { "指定 block file cache 在 BE 上的存储路径，默认 'random'，随机选择 BE 配置的存储路径。", @@ -5810,6 +5816,7 @@ public class SessionVariable implements Serializable, Writable { tResult.setSkipDeleteBitmap(skipDeleteBitmap); tResult.setEnableFileCache(enableFileCache); + tResult.setEnableFileCacheIoAdaptor(enableFileCacheIoAdaptor); tResult.setEnablePageCache(enablePageCache); diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableFileCacheIoAdaptorTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableFileCacheIoAdaptorTest.java new file mode 100644 index 00000000000..b208d2e1496 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableFileCacheIoAdaptorTest.java @@ -0,0 +1,51 @@ +// 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. + +package org.apache.doris.qe; + +import org.apache.doris.analysis.SetType; +import org.apache.doris.analysis.SetVar; +import org.apache.doris.analysis.StringLiteral; +import org.apache.doris.thrift.TQueryOptions; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class SessionVariableFileCacheIoAdaptorTest { + @Test + public void testDefaultIsExplicitlySentToBackend() { + TQueryOptions options = new SessionVariable().toThrift(); + Assertions.assertTrue(options.isSetEnableFileCacheIoAdaptor()); + Assertions.assertTrue(options.isEnableFileCacheIoAdaptor()); + } + + @Test + public void testSessionOverrideAndForwardingAreIsolated() throws Exception { + SessionVariable disabled = new SessionVariable(); + VariableMgr.setVar(disabled, new SetVar(SetType.SESSION, + SessionVariable.ENABLE_FILE_CACHE_IO_ADAPTOR, new StringLiteral("false"))); + TQueryOptions options = disabled.toThrift(); + Assertions.assertTrue(options.isSetEnableFileCacheIoAdaptor()); + Assertions.assertFalse(options.isEnableFileCacheIoAdaptor()); + Assertions.assertEquals("false", disabled.getForwardVariables() + .get(SessionVariable.ENABLE_FILE_CACHE_IO_ADAPTOR)); + SessionVariable forwarded = new SessionVariable(); + forwarded.setForwardedSessionVariables(disabled.getForwardVariables()); + Assertions.assertFalse(forwarded.toThrift().isEnableFileCacheIoAdaptor()); + Assertions.assertTrue(new SessionVariable().toThrift().isEnableFileCacheIoAdaptor()); + } +} diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift index 6c20c9087bb..e1f5a5303d0 100644 --- a/gensrc/thrift/PaloInternalService.thrift +++ b/gensrc/thrift/PaloInternalService.thrift @@ -514,6 +514,9 @@ struct TQueryOptions { // enable_paimon_cpp_reader. 229: optional bool enable_paimon_rust_reader = false; + // Explicitly sent by new FEs; an omitted field keeps older-FE queries on the local policy. + 230: optional bool enable_file_cache_io_adaptor = false; + // For cloud, to control if the content would be written into file cache // In write path, to control if the content would be written into file cache. // In read path, read from file cache or remote storage when execute query. --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
