github-actions[bot] commented on code in PR #69768:
URL: https://github.com/apache/doris/pull/69768#discussion_r4402098260
##########
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:
[P2] Preserve the inflight fast path before looking up disk blocks
This lookup always acquires `BlockFileCache::_mutex` before
`_read_async_write_path` can check the inflight index. With async writes, a
second read already covered by an inflight buffer now waits for unrelated cache
metadata/eviction work, including synchronous storage removal, although it
needs no disk or remote IO. Previously `_build_async_read_plan` returned on
`all_blocks_inflight` without taking this mutex. The default-enabled external
SELECT path therefore reintroduces cache-disk stalls into memory-only reads
even before the controller has samples. Check complete inflight coverage first,
and extend the existing mutex-held inflight regression to an enabled external
reader.
##########
be/test/io/cache/cached_remote_file_reader_test.cpp:
##########
@@ -1661,4 +1683,541 @@ 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);
Review Comment:
[P2] Give the internal test reader a valid tablet ID
`FileReaderOptions` defaults `tablet_id` to `-1`, but setting
`is_doris_table=true` makes the `CachedRemoteFileReader` constructor check
`_tablet_id > 0`. In a Debug/assertion-enabled BE test build this new test
aborts in the constructor before reaching `EXPECT_FALSE`, so it cannot validate
internal-reader exclusion and can terminate the test run. Set a positive
`tablet_id` before constructing the reader, as the existing internal-reader
setups do.
--
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]
