[feature](be) Reuse remote-only reads for no-write ranges

### What problem does this PR solve?

Issue Number: None

Related PR: #67292

Problem Summary: Asynchronous read-ahead ranges must preserve their coalesced byte boundaries and avoid populating File Cache before consumption is known. Route per-read NO_WRITE requests through CachedRemoteFileReader's remote-only-on-cache-miss path. Before probing downloaded cache blocks, reuse indexed in-flight async-write buffers when they completely cover the request, including requests spanning multiple cache blocks. If any in-flight block is absent, retain the existing all-cache-or-remote behavior: a fully downloaded range is served locally, while incomplete coverage issues one exact remote read for the entire planned range without creating cache cells, waiting for downloading blocks, reading from peers, or submitting cache writes. Keep cache-write mode resolution in the main reader dispatch because it selects among all read paths rather than belonging to the asynchronous cache-write implementation.

### Release note

None

### Check List (For Author)

- Test: Unit Test and BE build
    - ./run-be-ut.sh --run --filter='AsyncCachedRemoteFileReaderTest.exact_no_write*:BlockFileCacheTest.cached_remote_file_reader_remote_only_on_miss' -j100 (4 passed)
    - ./build.sh --be -j100
- Behavior changed: Yes; no-write asynchronous ranges reuse complete in-flight or downloaded-cache coverage and otherwise issue one exact remote read for the planned range.
- Does this need documentation: Yes; covered by the Phase 2 design document.

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/inflight_write_buffer_index.h"
 #include "io/cache/peer_file_cache_reader.h"
 #include "io/cache/remote_scan_cache_write_limiter.h"
 #include "io/fs/file_reader.h"
@@ -178,6 +179,24 @@ bool CachedRemoteFileReader::_should_read_from_peer(const IOContext* io_ctx) con
            doris::config::enable_cache_read_from_peer;
 }
 
+CacheWriteMode CachedRemoteFileReader::_resolve_cache_write_mode(const IOContext* io_ctx) const {
+    if (io_ctx->is_dryrun || io_ctx->is_warmup) {
+        return CacheWriteMode::SYNC_WRITE;
+    }
+    const auto mode = io_ctx->cache_write_mode_override.value_or(_cache_write_mode);
+    if (mode == CacheWriteMode::NO_WRITE) {
+        return mode;
+    }
+    if (_should_read_from_peer(io_ctx)) {
+        return CacheWriteMode::SYNC_WRITE;
+    }
+    if (mode != CacheWriteMode::DEFAULT) {
+        return mode;
+    }
+    return config::enable_async_file_cache_write ? CacheWriteMode::ASYNC_WRITE
+                                                 : CacheWriteMode::SYNC_WRITE;
+}
+
 void CachedRemoteFileReader::_insert_file_reader(FileBlockSPtr file_block) {
     if (_can_read_cache_file_directly()) {
         std::lock_guard lock(_mtx);
@@ -1130,6 +1149,53 @@ Status CachedRemoteFileReader::_read_from_indirect_cache(size_t offset, Slice re
     return Status::OK();
 }
 
+bool CachedRemoteFileReader::_try_read_from_inflight_buffers(size_t offset, Slice result,
+                                                             size_t bytes_req, bool is_dryrun,
+                                                             ReadStatistics& stats) {
+    if (!config::enable_async_file_cache_write_inflight_write_buffer_index) {
+        return false;
+    }
+    auto* inflight_index = _cache->inflight_write_buffer_index();
+    DORIS_CHECK(inflight_index != nullptr);
+    const size_t block_size = static_cast<size_t>(config::file_cache_each_block_size);
+    DORIS_CHECK(block_size > 0);
+    const size_t request_end = offset + bytes_req;
+    std::vector<size_t> block_offsets;
+    for (size_t block_offset = offset / block_size * block_size; block_offset < request_end;
+         block_offset += block_size) {
+        block_offsets.emplace_back(block_offset);
+    }
+    auto entries = inflight_index->lookup_all(_cache_hash, block_offsets);
+    bool fully_covered = true;
+    for (const auto& item : entries) {
+        if (item.entry == nullptr) {
+            ++stats.inflight_write_buffer_index_miss;
+            fully_covered = false;
+        } else {
+            ++stats.inflight_write_buffer_index_hit;
+        }
+    }
+    if (!fully_covered) {
+        return false;
+    }
+    for (const auto& item : entries) {
+        const auto& entry = item.entry;
+        DORIS_CHECK(entry->buffer != nullptr);
+        DORIS_CHECK(entry->buffer_offset == item.block_offset);
+        const size_t copy_left = std::max(offset, item.block_offset);
+        const size_t copy_end = std::min(request_end, item.block_offset + block_size);
+        const size_t copy_size = copy_end - copy_left;
+        const size_t buffer_offset = copy_left - item.block_offset;
+        DORIS_CHECK(buffer_offset <= entry->buffer_size);
+        DORIS_CHECK(copy_size <= entry->buffer_size - buffer_offset);
+        if (!is_dryrun) [[likely]] {
+            std::memcpy(result.data + copy_left - offset, entry->buffer->data() + buffer_offset,
+                        copy_size);
+        }
+    }
+    return true;
+}
+
 Status CachedRemoteFileReader::_read_remote_only_on_cache_miss(
         size_t offset, Slice result, size_t bytes_req, bool is_dryrun, size_t* bytes_read,
         ReadStatistics& stats, SourceReadBreakdown& source_read_breakdown,
@@ -1163,6 +1229,18 @@ Status CachedRemoteFileReader::_read_remote_only_on_cache_miss(
     };
 
     g_read_cache_indirect_num << 1;
+    if (_try_read_from_inflight_buffers(offset, result, bytes_req, is_dryrun, stats)) {
+        const size_t local_read_bytes = is_dryrun ? 0 : bytes_req;
+        if (is_dryrun) [[unlikely]] {
+            g_skip_local_cache_io_sum_bytes << bytes_req;
+        }
+        *bytes_read = bytes_req;
+        stats.hit_cache = true;
+        source_read_breakdown.local_bytes += local_read_bytes;
+        g_read_cache_indirect_bytes << local_read_bytes;
+        g_read_cache_indirect_total_bytes << bytes_req;
+        return Status::OK();
+    }
     CacheContext cache_context(io_ctx);
     cache_context.stats = &stats;
     cache_context.tablet_id = _tablet_id;
@@ -1291,7 +1369,9 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t*
         }
     }};
 
-    if (use_remote_only_on_cache_miss(io_ctx)) {
+    const CacheWriteMode cache_write_mode = _resolve_cache_write_mode(io_ctx);
+    TEST_SYNC_POINT("CachedRemoteFileReader::read_at_impl:after_resolve_cache_write_mode");
+    if (cache_write_mode == CacheWriteMode::NO_WRITE || 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);
         return read_st;
@@ -1303,8 +1383,6 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t*
         return Status::OK();
     }
 
-    const CacheWriteMode cache_write_mode = _resolve_cache_write_mode(io_ctx);
-    TEST_SYNC_POINT("CachedRemoteFileReader::read_at_impl:after_resolve_cache_write_mode");
     DORIS_CHECK(_cache_align_mode == CacheAlignMode::ALIGN_TO_BLOCK);
     DORIS_CHECK(cache_write_mode == CacheWriteMode::SYNC_WRITE ||
                 cache_write_mode == CacheWriteMode::ASYNC_WRITE);

be/src/io/cache/cached_remote_file_reader.h

@@ -149,7 +149,7 @@ private:
     /// Resolve the write policy for the current read instead of freezing a global setting in the
     /// reader constructor. Explicit cache-population reads always remain synchronous.
     /// @param[in] io_ctx Per-read flags and an optional write-mode override.
-    /// @return The effective synchronous or asynchronous cache-write mode for this read.
+    /// @return The effective no-write, synchronous, or asynchronous cache-write mode.
     CacheWriteMode _resolve_cache_write_mode(const IOContext* io_ctx) const;
 
     /// Serve a normal read while moving cache-miss writes off the query thread. The method uses
@@ -335,8 +335,13 @@ private:
                                      SourceReadBreakdown& source_read_breakdown,
                                      const IOContext* io_ctx);
 
-    /// Read local cache only when downloaded blocks fully cover the request; otherwise read remote
-    /// data directly without writing file cache.
+    /// Try to read the complete request from indexed async-write buffers. Return false without
+    /// modifying `result` when any covered cache block has no in-flight entry.
+    bool _try_read_from_inflight_buffers(size_t offset, Slice result, size_t bytes_req,
+                                         bool is_dryrun, ReadStatistics& stats);
+
+    /// Read in-flight buffers or downloaded cache blocks only when one source fully covers the
+    /// request; otherwise read remote data directly without writing file cache.
     /// @param[in] offset Original request offset.
     /// @param[out] result Destination buffer for the original request.
     /// @param[in] bytes_req Original request size.

be/src/io/cache/cached_remote_file_reader_async_write.cpp

@@ -168,21 +168,6 @@ struct CachedRemoteFileReader::AsyncReadPlan {
     }
 };
 
-// Resolve mode on every read so online configuration changes affect existing readers.
-CacheWriteMode CachedRemoteFileReader::_resolve_cache_write_mode(const IOContext* io_ctx) const {
-    if (io_ctx->is_dryrun || io_ctx->is_warmup || _should_read_from_peer(io_ctx)) {
-        return CacheWriteMode::SYNC_WRITE;
-    }
-    if (io_ctx->cache_write_mode_override.has_value()) {
-        return *io_ctx->cache_write_mode_override;
-    }
-    if (_cache_write_mode != CacheWriteMode::DEFAULT) {
-        return _cache_write_mode;
-    }
-    return config::enable_async_file_cache_write ? CacheWriteMode::ASYNC_WRITE
-                                                 : CacheWriteMode::SYNC_WRITE;
-}
-
 // Build a deliberately small plan. The inflight index is checked first so a fully covered read
 // avoids BlockFileCache::probe and its cache mutex. Only an incomplete inflight lookup performs one
 // whole-range probe whose result entries map directly to the logical plan blocks.

be/test/io/cache/cached_remote_file_reader_test.cpp

@@ -62,6 +62,7 @@ public:
     size_t last_offset() const { return _last_offset; }
     size_t last_size() const { return _last_size; }
     bool last_bypass_peer_read() const { return _last_bypass_peer_read; }
+    const std::vector<std::pair<size_t, size_t>>& read_ranges() const { return _read_ranges; }
 
 protected:
     Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
@@ -70,6 +71,7 @@ protected:
         _last_offset = offset;
         _last_size = result.size;
         _last_bypass_peer_read = io_ctx != nullptr && io_ctx->bypass_peer_read;
+        _read_ranges.emplace_back(offset, result.size);
         return _delegate->read_at(offset, result, bytes_read, io_ctx);
     }
 
@@ -79,6 +81,7 @@ private:
     size_t _last_offset {0};
     size_t _last_size {0};
     bool _last_bypass_peer_read {false};
+    std::vector<std::pair<size_t, size_t>> _read_ranges;
 };
 
 // Provides one isolated cache disk with phase-one async-read settings for concise end-to-end
@@ -1036,6 +1039,161 @@ TEST_F(AsyncCachedRemoteFileReaderTest,
     EXPECT_EQ(second_stats.async_cache_write_submitted, 0);
 }
 
+TEST_F(AsyncCachedRemoteFileReaderTest, exact_no_write_cold_miss_reads_only_the_requested_bytes) {
+    create_cache("cached_remote_reader_exact_cold_miss");
+    auto counting_reader = std::make_shared<CountingFileReader>(open_remote_file());
+    auto reader = create_reader(counting_reader);
+
+    constexpr size_t read_offset = 12345;
+    constexpr size_t read_size = 64_kb;
+    std::string result(read_size, '\0');
+    FileCacheStatistics stats;
+    IOContext context;
+    context.file_cache_stats = &stats;
+    context.cache_write_mode_override = CacheWriteMode::NO_WRITE;
+    size_t bytes_read = 0;
+    ASSERT_TRUE(
+            reader->read_at(read_offset, Slice(result.data(), result.size()), &bytes_read, &context)
+                    .ok());
+
+    EXPECT_EQ(bytes_read, read_size);
+    EXPECT_EQ(result, std::string(read_size, '0'));
+    ASSERT_EQ(counting_reader->read_ranges().size(), 1);
+    EXPECT_EQ(counting_reader->read_ranges().front(), std::make_pair(read_offset, read_size));
+    EXPECT_EQ(stats.bytes_read_from_remote, read_size);
+    EXPECT_EQ(stats.bytes_read_from_local, 0);
+    EXPECT_EQ(stats.bytes_read_from_peer, 0);
+    EXPECT_EQ(stats.bytes_write_into_cache, 0);
+    EXPECT_EQ(stats.num_skip_cache_io_total, 1);
+    EXPECT_EQ(cache()->async_write_manager()->pending_count(), 0);
+
+    ReadStatistics probe_stats;
+    CacheContext probe_context;
+    probe_context.stats = &probe_stats;
+    auto probe_result = cache()->probe(reader->_cache_hash, 0, 1_mb, probe_context);
+    ASSERT_EQ(probe_result.file_blocks.size(), 1);
+    EXPECT_EQ(probe_result.file_blocks.front(), nullptr);
+}
+
+TEST_F(AsyncCachedRemoteFileReaderTest,
+       exact_no_write_reuses_only_complete_multiblock_inflight_coverage) {
+    create_cache("cached_remote_reader_exact_inflight_coverage");
+    auto counting_reader = std::make_shared<CountingFileReader>(open_remote_file());
+    auto reader = create_reader(counting_reader);
+    auto* manager = cache()->async_write_manager();
+    auto* index = cache()->inflight_write_buffer_index();
+
+    AsyncCacheWriteBufferPtr first_buffer;
+    ASSERT_TRUE(manager->allocate_tracked_buffer(1_mb, &first_buffer).ok());
+    std::memset(first_buffer->data(), 'a', first_buffer->size());
+    auto first_entry =
+            std::make_shared<InflightWriteBufferEntry>(first_buffer, 0, 1_mb, MonotonicMicros());
+    ASSERT_EQ(index->insert_if_absent(reader->_cache_hash, 0, first_entry), nullptr);
+    Defer remove_first {
+            [&]() { static_cast<void>(index->remove_if(reader->_cache_hash, 0, first_entry)); }};
+
+    AsyncCacheWriteBufferPtr second_buffer;
+    ASSERT_TRUE(manager->allocate_tracked_buffer(1_mb, &second_buffer).ok());
+    std::memset(second_buffer->data(), 'b', second_buffer->size());
+    auto second_entry = std::make_shared<InflightWriteBufferEntry>(second_buffer, 1_mb, 1_mb,
+                                                                   MonotonicMicros());
+    ASSERT_EQ(index->insert_if_absent(reader->_cache_hash, 1_mb, second_entry), nullptr);
+
+    IOContext context;
+    context.cache_write_mode_override = CacheWriteMode::NO_WRITE;
+    constexpr size_t covered_offset = 1_mb - 64_kb;
+    constexpr size_t covered_size = 128_kb;
+    std::string fully_covered_result(covered_size, '\0');
+    FileCacheStatistics fully_covered_stats;
+    context.file_cache_stats = &fully_covered_stats;
+    size_t bytes_read = 0;
+    ASSERT_TRUE(reader->read_at(covered_offset,
+                                Slice(fully_covered_result.data(), fully_covered_result.size()),
+                                &bytes_read, &context)
+                        .ok());
+
+    EXPECT_EQ(bytes_read, covered_size);
+    EXPECT_EQ(fully_covered_result, std::string(64_kb, 'a') + std::string(64_kb, 'b'));
+    EXPECT_EQ(counting_reader->read_count(), 0);
+    EXPECT_EQ(fully_covered_stats.bytes_read_from_local, covered_size);
+    EXPECT_EQ(fully_covered_stats.bytes_read_from_remote, 0);
+    EXPECT_EQ(fully_covered_stats.inflight_write_buffer_index_hit, 2);
+    EXPECT_EQ(fully_covered_stats.inflight_write_buffer_index_miss, 0);
+
+    ASSERT_TRUE(index->remove_if(reader->_cache_hash, 1_mb, second_entry));
+    constexpr size_t partial_offset = covered_offset;
+    constexpr size_t partial_size = covered_size;
+    std::string partially_covered_result(partial_size, '\0');
+    FileCacheStatistics partially_covered_stats;
+    context.file_cache_stats = &partially_covered_stats;
+    bytes_read = 0;
+    ASSERT_TRUE(
+            reader->read_at(partial_offset,
+                            Slice(partially_covered_result.data(), partially_covered_result.size()),
+                            &bytes_read, &context)
+                    .ok());
+
+    EXPECT_EQ(bytes_read, partial_size);
+    EXPECT_EQ(partially_covered_result, std::string(64_kb, '0') + std::string(64_kb, '1'));
+    ASSERT_EQ(counting_reader->read_ranges().size(), 1);
+    EXPECT_EQ(counting_reader->read_ranges().front(), std::make_pair(partial_offset, partial_size));
+    EXPECT_EQ(partially_covered_stats.bytes_read_from_local, 0);
+    EXPECT_EQ(partially_covered_stats.bytes_read_from_remote, partial_size);
+    EXPECT_EQ(partially_covered_stats.inflight_write_buffer_index_hit, 1);
+    EXPECT_EQ(partially_covered_stats.inflight_write_buffer_index_miss, 1);
+}
+
+TEST_F(AsyncCachedRemoteFileReaderTest,
+       exact_no_write_partial_cache_coverage_uses_one_remote_read) {
+    create_cache("cached_remote_reader_exact_partial_hit");
+    auto counting_reader = std::make_shared<CountingFileReader>(open_remote_file());
+    auto reader = create_reader(counting_reader);
+
+    ReadStatistics cache_stats;
+    CacheContext cache_context;
+    cache_context.stats = &cache_stats;
+    auto holder = cache()->get_or_set(reader->_cache_hash, 1_mb, 1_mb, cache_context);
+    ASSERT_EQ(holder.file_blocks.size(), 1);
+    const auto& cached_block = holder.file_blocks.front();
+    ASSERT_EQ(cached_block->get_or_set_downloader(), FileBlock::get_caller_id());
+    const std::string cached_payload(1_mb, 'x');
+    ASSERT_TRUE(cached_block->append(Slice(cached_payload.data(), cached_payload.size())).ok());
+    ASSERT_TRUE(cached_block->finalize().ok());
+
+    constexpr size_t gap_size = 64_kb;
+    constexpr size_t read_offset = 1_mb - gap_size;
+    constexpr size_t read_size = 1_mb + 2 * gap_size;
+    std::string result(read_size, '\0');
+    FileCacheStatistics stats;
+    IOContext context;
+    context.file_cache_stats = &stats;
+    context.cache_write_mode_override = CacheWriteMode::NO_WRITE;
+    size_t bytes_read = 0;
+    ASSERT_TRUE(
+            reader->read_at(read_offset, Slice(result.data(), result.size()), &bytes_read, &context)
+                    .ok());
+
+    EXPECT_EQ(bytes_read, read_size);
+    EXPECT_EQ(result.substr(0, gap_size), std::string(gap_size, '0'));
+    EXPECT_EQ(result.substr(gap_size, 1_mb), std::string(1_mb, '1'));
+    EXPECT_EQ(result.substr(gap_size + 1_mb), std::string(gap_size, '2'));
+    ASSERT_EQ(counting_reader->read_ranges().size(), 1);
+    EXPECT_EQ(counting_reader->read_ranges().front(), std::make_pair(read_offset, read_size));
+    EXPECT_EQ(stats.bytes_read_from_local, 0);
+    EXPECT_EQ(stats.bytes_read_from_remote, read_size);
+    EXPECT_EQ(stats.bytes_write_into_cache, 0);
+    EXPECT_EQ(cache()->async_write_manager()->pending_count(), 0);
+
+    ReadStatistics probe_stats;
+    CacheContext probe_context;
+    probe_context.stats = &probe_stats;
+    auto probe_result = cache()->probe(reader->_cache_hash, 0, 3_mb, probe_context);
+    ASSERT_EQ(probe_result.file_blocks.size(), 3);
+    EXPECT_EQ(probe_result.file_blocks[0], nullptr);
+    EXPECT_EQ(probe_result.file_blocks[1], cached_block);
+    EXPECT_EQ(probe_result.file_blocks[2], nullptr);
+}
+
 TEST_F(BlockFileCacheTest,
        direct_partial_hit_with_downloaded_remainder_should_not_read_remote_again) {
     std::string local_cache_base_path =
@@ -1708,6 +1866,9 @@ TEST_F(BlockFileCacheTest, cache_write_mode_is_resolved_for_each_read_context) {
     config::deploy_mode = "cloud";
     config::enable_cache_read_from_peer = true;
     EXPECT_EQ(reader->_resolve_cache_write_mode(&context), CacheWriteMode::SYNC_WRITE);
+    context.cache_write_mode_override = CacheWriteMode::NO_WRITE;
+    EXPECT_EQ(reader->_resolve_cache_write_mode(&context), CacheWriteMode::NO_WRITE);
+    context.cache_write_mode_override = CacheWriteMode::ASYNC_WRITE;
     context.bypass_peer_read = true;
     EXPECT_EQ(reader->_resolve_cache_write_mode(&context), CacheWriteMode::ASYNC_WRITE);