[feature](be) Connect read-ahead range writeback

### What problem does this PR solve?

Issue Number: Part of #67611

Related PR: #67292, #67579

Problem Summary: Centralize query read-ahead construction so regular scans and exact-row callers share range planning, byte windows, cache probes, scheduler ownership and writeback setup. Merge per-read IO statistics into caller accounting and publish each consumed range once. Submit writeback only for consumed ranges that read remote bytes, preserving no-write policy and avoiding redundant hole filling for File Cache or inflight hits. Carry the read identity through the consumed-range callback so offline traces can attribute background work to its foreground source.

Require both query read-ahead and asynchronous cache writes so disabling the writer preserves synchronous cache filling. Handle a switch changing during scan initialization through the original reader. Keep delayed hole-fill sources open with an aliasing shared_ptr that retains the cached wrapper; regression tests release both the foreground reader and writeback context before completing the queued fill.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - Component tests are included with the implementation; rebase and synchronization validation is recorded in the related PRs and tracking issue.
    - Repository clang-format 16, header hygiene and git diff --check passed before folding.
    - No cluster, external-object-store benchmark or clang-tidy run during synchronization.
- Behavior changed: Yes, includes final component behavior and applicable diagnostic instrumentation; default enable switches remain off.
- Does this need documentation: Yes, tools/read_io_trace/README.md and related PR/tracking descriptions.

be/src/io/cache/cached_remote_file_reader.cpp

@@ -50,6 +50,7 @@
 #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/range_cache_writeback.h"
 #include "io/cache/remote_scan_cache_write_limiter.h"
 #include "io/fs/file_reader.h"
 #include "io/fs/local_file_system.h"
@@ -135,6 +136,34 @@ CachedRemoteFileReader::CachedRemoteFileReader(FileReaderSPtr remote_file_reader
     }
 }
 
+std::shared_ptr<RangeCacheWriteback> CachedRemoteFileReader::make_range_cache_writeback(
+        const IOContext& io_context, PartialBlockWritebackManager* partial_block_manager) const {
+    DORIS_CHECK(_cache != nullptr);
+    DORIS_CHECK(partial_block_manager != nullptr);
+    if (_resolve_cache_write_mode(&io_context) == CacheWriteMode::NO_WRITE ||
+        use_remote_only_on_cache_miss(&io_context)) {
+        return nullptr;
+    }
+    auto* write_manager = _cache->async_write_manager();
+    DORIS_CHECK(write_manager != nullptr);
+    CacheContext cache_context(&io_context);
+    return std::make_shared<RangeCacheWriteback>(RangeCacheWritebackOptions {
+            .write_manager = write_manager,
+            .partial_block_manager = partial_block_manager,
+            .inflight_index = config::enable_async_file_cache_write_inflight_write_buffer_index
+                                      ? _cache->inflight_write_buffer_index()
+                                      : nullptr,
+            // The wrapper closes the inner reader on destruction. Keep the wrapper as the owner
+            // while background hole fills still read directly through the inner reader.
+            .source_reader = FileReaderSPtr(shared_from_this(), _remote_file_reader.get()),
+            .cache_hash = _cache_hash,
+            .file_size = size(),
+            .block_size = static_cast<size_t>(config::file_cache_each_block_size),
+            .admission_ctx = CacheAdmissionContext::from_cache_context(cache_context, _tablet_id),
+            .io_context = FileRangeReadIOContext::from_caller(io_context),
+    });
+}
+
 void CachedRemoteFileReader::_init_doris_table_cache() {
     _cache_hash = BlockFileCache::hash(path().filename().native());
     _cache = FileCacheFactory::instance()->get_by_path(_cache_hash);

be/src/io/cache/cached_remote_file_reader.h

@@ -39,6 +39,8 @@ struct AsyncCacheWriteEpoch;
 struct IOContext;
 struct FileCacheStatistics;
 struct PeerFetchResult;
+class PartialBlockWritebackManager;
+class RangeCacheWriteback;
 
 } // namespace doris::io
 
@@ -87,6 +89,13 @@ public:
     /// @return Raw pointer to the underlying reader owned by this object.
     FileReader* get_remote_reader() { return _remote_file_reader.get(); }
 
+    /// Build the immutable cache metadata used to write back consumed query read-ahead ranges.
+    /// The returned object reads directly from the inner reader while retaining this wrapper,
+    /// keeping its destructor from closing the source before background hole fills finish.
+    /// A no-write request returns nullptr.
+    std::shared_ptr<RangeCacheWriteback> make_range_cache_writeback(
+            const IOContext& io_context, PartialBlockWritebackManager* partial_block_manager) const;
+
     /// Align a read range to file-cache block boundaries.
     /// @param[in] offset Requested read offset in bytes.
     /// @param[in] size Requested read size in bytes.

be/src/io/io_common.h

@@ -59,6 +59,13 @@ struct FileReaderStats {
     size_t read_bytes = 0;
     int64_t read_time_ns = 0;
     size_t read_rows = 0;
+
+    void merge_from(const FileReaderStats& other) {
+        read_calls += other.read_calls;
+        read_bytes += other.read_bytes;
+        read_time_ns += other.read_time_ns;
+        read_rows += other.read_rows;
+    }
 };
 
 struct FileCacheStatistics {

be/src/storage/segment/segment_iterator.cpp

@@ -68,7 +68,6 @@
 #include "exprs/vliteral.h"
 #include "exprs/vmatch_predicate.h"
 #include "exprs/vslot_ref.h"
-#include "io/cache/cached_remote_file_reader.h"
 #include "io/fs/file_reader.h"
 #include "io/io_common.h"
 #include "runtime/exec_env.h"
@@ -387,8 +386,7 @@ Status SegmentIterator::init(const StorageReadOptions& opts) {
 }
 
 Status SegmentIterator::_init_query_read_ahead() {
-    if (!config::enable_query_read_ahead || !config::is_cloud_mode() ||
-        _opts.io_ctx.reader_type != ReaderType::READER_QUERY) {
+    if (!SegmentReadAhead::enabled() || _opts.io_ctx.reader_type != ReaderType::READER_QUERY) {
         return Status::OK();
     }
 
@@ -399,44 +397,21 @@ Status SegmentIterator::_init_query_read_ahead() {
     DORIS_CHECK(scheduler != nullptr);
     auto* query_context = _opts.runtime_state->get_query_ctx();
     DORIS_CHECK(query_context != nullptr);
-
-    ColumnReadAheadOptions eager_options {
-            .window_bytes = cast_set<size_t>(config::read_ahead_eager_window_bytes),
-    };
-    ColumnReadAheadOptions lazy_options {
-            .window_bytes = cast_set<size_t>(config::read_ahead_lazy_window_bytes),
-    };
-    const auto eager_status = eager_options.validate();
-    const auto lazy_status = lazy_options.validate();
-    if (!eager_status.ok() || !lazy_status.ok()) {
-        LOG_EVERY_N(WARNING, 100) << "invalid query read-ahead byte windows; feature disabled for "
-                                     "this segment: eager="
-                                  << eager_status << ", lazy=" << lazy_status;
+    auto read_context = query_context->get_or_create_file_range_read_context(scheduler);
+    const auto status = SegmentReadAhead::create_for_query(
+            _file_reader, exec_env, std::move(read_context), _opts, &_segment_read_ahead);
+    if (!status.ok()) {
+        LOG_EVERY_N(WARNING, 100)
+                << "failed to initialize query read-ahead; use the original read path: " << status;
         return Status::OK();
     }
-    SegmentReadAheadOptions options {
-            .range_plan = {.coalesce_options =
-                                   {.max_gap_bytes =
-                                            cast_set<size_t>(config::read_ahead_max_gap_bytes),
-                                    .max_range_bytes =
-                                            cast_set<size_t>(config::read_ahead_max_range_bytes),
-                                    .max_read_amplification_ratio =
-                                            config::read_ahead_max_read_amplification_ratio},
-                           .cache_block_size = cast_set<size_t>(config::file_cache_each_block_size),
-                           .block_fill_min_coverage = config::read_ahead_block_fill_min_coverage},
-            .page_cache_probe = _opts.use_page_cache ? make_storage_page_cache_probe(_file_reader)
-                                                     : ReadAheadPageCacheProbe {},
-            .range_consumer_factory = {},
-    };
-    auto read_context = query_context->get_or_create_file_range_read_context(scheduler);
-    _segment_read_ahead = std::make_unique<SegmentReadAhead>(
-            _file_reader, scheduler, std::move(read_context),
-            io::FileRangeReadIOContext::from_caller(_opts.io_ctx), std::move(options));
-    _column_read_ahead_context = std::make_unique<ColumnReadAheadContext>(ColumnReadAheadContext {
-            .eager_options = eager_options,
-            .lazy_options = lazy_options,
-            .segment = _segment_read_ahead.get(),
-    });
+    // Both enable switches are mutable. If one was disabled after the entry check,
+    // create_for_query() leaves the context empty and the original reader remains in use.
+    if (_segment_read_ahead == nullptr) {
+        return Status::OK();
+    }
+    _column_read_ahead_context =
+            std::make_unique<ColumnReadAheadContext>(_segment_read_ahead->column_context());
     return Status::OK();
 }
 

be/src/storage/segment/segment_read_ahead.cpp

@@ -23,9 +23,16 @@
 #include <cstring>
 #include <utility>
 
+#include "common/cast_set.h"
+#include "common/config.h"
 #include "common/logging.h"
+#include "io/cache/cached_remote_file_reader.h"
+#include "io/cache/range_cache_writeback.h"
 #include "io/fs/read_ahead_metrics.h"
+#include "runtime/exec_env.h"
 #include "storage/cache/page_cache.h"
+#include "storage/iterators.h"
+#include "storage/segment/column_reader.h"
 
 namespace doris::segment_v2 {
 namespace {
@@ -156,7 +163,7 @@ void SegmentReadAheadFileReader::_finish_page(const PageKey& key, bool consumed)
     }
     if (consumed_range != nullptr && consumed_range->consumer) {
         auto consumer = std::move(consumed_range->consumer);
-        consumer(consumed_range->read->range(), consumed_range->read->data());
+        consumer(*consumed_range->read);
     }
 }
 
@@ -182,7 +189,9 @@ SegmentReadAhead::SegmentReadAhead(io::FileReaderSPtr source_reader,
                                    io::FileRangeReadScheduler* scheduler,
                                    std::shared_ptr<io::FileRangeReadContext> context,
                                    io::FileRangeReadIOContext io_context,
-                                   SegmentReadAheadOptions options)
+                                   SegmentReadAheadOptions options,
+                                   ColumnReadAheadOptions eager_options,
+                                   ColumnReadAheadOptions lazy_options)
         : _source_reader(std::move(source_reader)),
           _scheduler(scheduler),
           _context(std::move(context)),
@@ -192,13 +201,115 @@ SegmentReadAhead::SegmentReadAhead(io::FileReaderSPtr source_reader,
                               ? _io_context.statistics
                               : std::make_shared<io::ReadAheadStatistics>()),
           _reader(std::shared_ptr<SegmentReadAheadFileReader>(
-                  new SegmentReadAheadFileReader(_source_reader, _statistics))) {
+                  new SegmentReadAheadFileReader(_source_reader, _statistics))),
+          _column_context {
+                  .eager_options = eager_options, .lazy_options = lazy_options, .segment = this} {
     DORIS_CHECK(_source_reader != nullptr);
     DORIS_CHECK(_scheduler != nullptr);
     DORIS_CHECK(_context != nullptr);
     _io_context.statistics = _statistics;
 }
 
+bool SegmentReadAhead::enabled() {
+    // Read-ahead fetches with NO_WRITE and relies on the async writer after consumption.
+    // With only read-ahead enabled, retain the original reader so cold pages still fill cache.
+    return config::enable_query_read_ahead && config::enable_async_file_cache_write &&
+           config::is_cloud_mode();
+}
+
+Status SegmentReadAhead::create_for_query(io::FileReaderSPtr source_reader, ExecEnv* exec_env,
+                                          std::shared_ptr<io::FileRangeReadContext> context,
+                                          const StorageReadOptions& read_options,
+                                          std::unique_ptr<SegmentReadAhead>* output) {
+    DORIS_CHECK(source_reader != nullptr);
+    DORIS_CHECK(exec_env != nullptr);
+    DORIS_CHECK(output != nullptr);
+    output->reset();
+    if (!enabled() || read_options.io_ctx.reader_type != ReaderType::READER_QUERY) {
+        return Status::OK();
+    }
+    DORIS_CHECK(context != nullptr);
+
+    ColumnReadAheadOptions eager_options {
+            .window_bytes = cast_set<size_t>(config::read_ahead_eager_window_bytes),
+    };
+    ColumnReadAheadOptions lazy_options {
+            .window_bytes = cast_set<size_t>(config::read_ahead_lazy_window_bytes),
+    };
+    RETURN_IF_ERROR(eager_options.validate());
+    RETURN_IF_ERROR(lazy_options.validate());
+
+    auto statistics =
+            read_options.stats != nullptr ? read_options.stats->read_ahead_stats : nullptr;
+    if (statistics == nullptr) {
+        statistics = std::make_shared<io::ReadAheadStatistics>();
+        if (read_options.stats != nullptr) {
+            read_options.stats->read_ahead_stats = statistics;
+        }
+    }
+
+    auto* file_cache_stats = read_options.io_ctx.file_cache_stats;
+    auto* file_reader_stats = read_options.io_ctx.file_reader_stats;
+    std::shared_ptr<io::RangeCacheWriteback> range_writeback;
+    auto cached_reader = std::dynamic_pointer_cast<io::CachedRemoteFileReader>(source_reader);
+    if (cached_reader != nullptr) {
+        auto* partial_block_manager = exec_env->partial_block_writeback_manager();
+        DORIS_CHECK(partial_block_manager != nullptr);
+        range_writeback = cached_reader->make_range_cache_writeback(read_options.io_ctx,
+                                                                    partial_block_manager);
+    }
+    ReadAheadRangeConsumerFactory range_consumer_factory;
+    if (file_cache_stats != nullptr || file_reader_stats != nullptr || range_writeback != nullptr) {
+        range_consumer_factory = [file_cache_stats, file_reader_stats, statistics,
+                                  range_writeback]() -> ReadAheadRangeConsumer {
+            std::optional<io::AsyncCacheWriteEpoch> write_epoch;
+            if (range_writeback != nullptr) {
+                write_epoch = range_writeback->capture_write_epoch();
+            }
+            return [file_cache_stats, file_reader_stats, range_writeback, statistics,
+                    write_epoch = std::move(write_epoch)](const io::FileRangeRead& read) {
+                const auto stats = read.stats();
+                if (file_cache_stats != nullptr) {
+                    file_cache_stats->merge_from(stats.file_cache);
+                }
+                if (file_reader_stats != nullptr) {
+                    file_reader_stats->merge_from(stats.file_reader);
+                }
+                // Cache and inflight-buffer hits already have a complete cache-block owner.
+                if (write_epoch.has_value() && stats.file_cache.bytes_read_from_remote > 0) {
+                    static_cast<void>(range_writeback->submit_consumed_range(
+                            read.range(), read.data(), *write_epoch, statistics.get(),
+                            read.trace_id()));
+                }
+            };
+        };
+    }
+
+    SegmentReadAheadOptions options {
+            .range_plan = {.coalesce_options =
+                                   {.max_gap_bytes =
+                                            cast_set<size_t>(config::read_ahead_max_gap_bytes),
+                                    .max_range_bytes =
+                                            cast_set<size_t>(config::read_ahead_max_range_bytes),
+                                    .max_read_amplification_ratio =
+                                            config::read_ahead_max_read_amplification_ratio},
+                           .cache_block_size = cast_set<size_t>(config::file_cache_each_block_size),
+                           .block_fill_min_coverage = config::read_ahead_block_fill_min_coverage},
+            .page_cache_probe = read_options.use_page_cache
+                                        ? make_storage_page_cache_probe(source_reader)
+                                        : ReadAheadPageCacheProbe {},
+            .range_consumer_factory = std::move(range_consumer_factory),
+    };
+    auto* scheduler = exec_env->file_range_read_scheduler();
+    DORIS_CHECK(scheduler != nullptr);
+    auto io_context = io::FileRangeReadIOContext::from_caller(read_options.io_ctx);
+    io_context.statistics = std::move(statistics);
+    output->reset(new SegmentReadAhead(std::move(source_reader), scheduler, std::move(context),
+                                       std::move(io_context), std::move(options), eager_options,
+                                       lazy_options));
+    return Status::OK();
+}
+
 SegmentReadAheadResult SegmentReadAhead::apply_plans(std::vector<ColumnReadAheadPlan> plans) {
     COUNTER_UPDATE(&_statistics->plan_calls, 1);
     SCOPED_TIMER(&_statistics->plan_time);

be/src/storage/segment/segment_read_ahead.h

@@ -29,11 +29,18 @@
 #include "io/fs/file_reader.h"
 #include "storage/segment/column_read_ahead.h"
 
-namespace doris::segment_v2 {
+namespace doris {
+
+class ExecEnv;
+class StorageReadOptions;
+
+namespace segment_v2 {
+
+class ColumnIterator;
 
 /// Invoked with the whole physical range at most once, when any registered page successfully
 /// consumes bytes from it.
-using ReadAheadRangeConsumer = std::function<void(const io::FileRange&, Slice)>;
+using ReadAheadRangeConsumer = std::function<void(const io::FileRangeRead&)>;
 /// Creates submission-scoped state immediately before asynchronous range tasks are admitted.
 using ReadAheadRangeConsumerFactory = std::function<ReadAheadRangeConsumer()>;
 /// Returns true when the exact compressed data-page range is already in StoragePageCache.
@@ -149,7 +156,19 @@ class SegmentReadAhead final {
 public:
     SegmentReadAhead(io::FileReaderSPtr source_reader, io::FileRangeReadScheduler* scheduler,
                      std::shared_ptr<io::FileRangeReadContext> context,
-                     io::FileRangeReadIOContext io_context, SegmentReadAheadOptions options);
+                     io::FileRangeReadIOContext io_context, SegmentReadAheadOptions options,
+                     ColumnReadAheadOptions eager_options = {},
+                     ColumnReadAheadOptions lazy_options = {});
+
+    /// Read the dynamic query read-ahead switch and cloud-mode requirement.
+    static bool enabled();
+
+    /// Create query read-ahead state when enabled for a query reader. On an inapplicable request,
+    /// return OK with `output` cleared so callers continue with the original reader.
+    static Status create_for_query(io::FileReaderSPtr source_reader, ExecEnv* exec_env,
+                                   std::shared_ptr<io::FileRangeReadContext> context,
+                                   const StorageReadOptions& read_options,
+                                   std::unique_ptr<SegmentReadAhead>* output);
 
     SegmentReadAhead(const SegmentReadAhead&) = delete;
     SegmentReadAhead& operator=(const SegmentReadAhead&) = delete;
@@ -161,6 +180,8 @@ public:
 
     const std::shared_ptr<SegmentReadAheadFileReader>& file_reader() const { return _reader; }
 
+    const ColumnReadAheadContext& column_context() const { return _column_context; }
+
 private:
     const io::FileReaderSPtr _source_reader;
     io::FileRangeReadScheduler* const _scheduler;
@@ -169,8 +190,10 @@ private:
     const SegmentReadAheadOptions _options;
     const std::shared_ptr<io::ReadAheadStatistics> _statistics;
     const std::shared_ptr<SegmentReadAheadFileReader> _reader;
+    const ColumnReadAheadContext _column_context;
 };
 
 ReadAheadPageCacheProbe make_storage_page_cache_probe(const io::FileReaderSPtr& file_reader);
 
-} // namespace doris::segment_v2
+} // namespace segment_v2
+} // namespace doris

be/test/io/cache/cached_remote_file_reader_test.cpp

@@ -22,8 +22,13 @@
 #include "cloud/config.h"
 #include "io/cache/async_cache_write_manager_metrics.h"
 #include "io/cache/inflight_write_buffer_index.h"
+#include "io/cache/partial_block_writeback_manager.h"
+#include "io/cache/range_cache_writeback.h"
 #include "io/file_factory.h"
+#include "io/fs/read_ahead_metrics.h"
 #include "io/fs/read_io_trace_test_util.h"
+#include "storage/iterators.h"
+#include "storage/segment/segment_read_ahead.h"
 #include "util/time.h"
 
 namespace doris::io {
@@ -351,6 +356,240 @@ TEST_F(AsyncCachedRemoteFileReaderTest, preallocated_cache_block_can_cover_the_s
     EXPECT_EQ(cached_stats.async_cache_write_submitted, 0);
 }
 
+TEST_F(AsyncCachedRemoteFileReaderTest, builds_consumed_range_writeback_context) {
+    create_cache("cached_remote_reader_range_writeback_context");
+    auto counting_reader = std::make_shared<CountingFileReader>(open_remote_file());
+    auto reader = create_reader(counting_reader);
+    std::unique_ptr<PartialBlockWritebackManager> partial_manager;
+    ASSERT_TRUE(PartialBlockWritebackManager::create(
+                        {.block_size = 1_mb,
+                         .worker_count = 1,
+                         .max_pending_bytes = 2_mb,
+                         .hole_fill_coalesce = {.max_gap_bytes = 32_kb,
+                                                .max_range_bytes = 1_mb,
+                                                .max_read_amplification_ratio = 2.0}},
+                        &partial_manager)
+                        .ok());
+    TUniqueId query_id;
+    query_id.hi = 123;
+    query_id.lo = 456;
+    IOContext io_context;
+    io_context.query_id = &query_id;
+    io_context.expiration_time = 789;
+    io_context.is_warmup = true;
+
+    auto writeback = reader->make_range_cache_writeback(io_context, partial_manager.get());
+
+    ASSERT_NE(writeback, nullptr);
+    EXPECT_EQ(writeback->_options.write_manager, cache()->async_write_manager());
+    EXPECT_EQ(writeback->_options.partial_block_manager, partial_manager.get());
+    EXPECT_EQ(writeback->_options.inflight_index, cache()->inflight_write_buffer_index());
+    EXPECT_EQ(writeback->_options.source_reader, counting_reader);
+    EXPECT_EQ(writeback->_options.cache_hash, reader->_cache_hash);
+    EXPECT_EQ(writeback->_options.file_size, reader->size());
+    EXPECT_EQ(writeback->_options.block_size, 1_mb);
+    EXPECT_EQ(writeback->_options.admission_ctx.query_id, query_id);
+    EXPECT_EQ(writeback->_options.admission_ctx.cache_type, FileCacheType::TTL);
+    EXPECT_EQ(writeback->_options.admission_ctx.expiration_time, 789);
+    EXPECT_EQ(writeback->_options.admission_ctx.tablet_id, 10086);
+    EXPECT_TRUE(writeback->_options.admission_ctx.is_warmup);
+    ASSERT_NE(writeback->_options.io_context.query_id, nullptr);
+    EXPECT_EQ(*writeback->_options.io_context.query_id, query_id);
+    EXPECT_NE(writeback->_options.io_context.io_context.query_id, &query_id);
+    EXPECT_EQ(writeback->_options.io_context.io_context.cache_write_mode_override,
+              CacheWriteMode::NO_WRITE);
+}
+
+TEST_F(AsyncCachedRemoteFileReaderTest, delayed_hole_fill_keeps_cached_reader_open) {
+    create_cache("delayed_hole_fill_reader_lifetime");
+    auto source = std::make_shared<CountingFileReader>(open_remote_file());
+    auto reader = create_reader(source);
+    const auto hash = reader->_cache_hash;
+    std::weak_ptr<CachedRemoteFileReader> weak_reader = reader;
+    std::unique_ptr<PartialBlockWritebackManager> partial_manager;
+    ASSERT_TRUE(PartialBlockWritebackManager::create(
+                        {.block_size = 1_mb,
+                         .worker_count = 1,
+                         .merge_delay_ms = 60000,
+                         .max_pending_bytes = 2_mb,
+                         .hole_fill_coalesce = {.max_gap_bytes = 32_kb,
+                                                .max_range_bytes = 1_mb,
+                                                .max_read_amplification_ratio = 1.0}},
+                        &partial_manager)
+                        .ok());
+    auto writeback = reader->make_range_cache_writeback(IOContext {}, partial_manager.get());
+    auto epoch = writeback->capture_write_epoch();
+    ASSERT_TRUE(epoch.has_value());
+    std::string fragment(16, '0');
+    static_cast<void>(writeback->submit_consumed_range({.offset = 0, .size = fragment.size()},
+                                                       Slice(fragment), *epoch));
+    ASSERT_EQ(partial_manager->queued_count(), 1);
+
+    writeback.reset();
+    reader.reset();
+    ASSERT_FALSE(weak_reader.expired());
+    ASSERT_FALSE(source->closed());
+
+    partial_manager->set_merge_delay_ms(0);
+    for (int attempt = 0; attempt < 5000 && !source->closed(); ++attempt) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    ASSERT_TRUE(weak_reader.expired());
+    wait_for_async_writes();
+    EXPECT_TRUE(source->closed());
+    ReadStatistics stats;
+    CacheContext context;
+    context.stats = &stats;
+    FileBlocks blocks;
+    bool covered = false;
+    ASSERT_TRUE(cache()->get_downloaded_blocks_if_fully_covered(hash, 0, 1_mb, context, &blocks,
+                                                                &covered)
+                        .ok());
+    ASSERT_TRUE(covered);
+    ASSERT_EQ(blocks.size(), 1);
+    std::string data(1_mb, '\0');
+    ASSERT_TRUE(blocks.front()->read(Slice(data), 0).ok());
+    EXPECT_EQ(data, std::string(1_mb, '0'));
+}
+
+TEST_F(AsyncCachedRemoteFileReaderTest, read_ahead_requires_async_cache_writes) {
+    const auto old_deploy_mode = config::deploy_mode;
+    const bool old_read_ahead = config::enable_query_read_ahead;
+    Defer restore {[&]() {
+        config::deploy_mode = old_deploy_mode;
+        config::enable_query_read_ahead = old_read_ahead;
+    }};
+    config::deploy_mode = "cloud";
+    config::enable_query_read_ahead = true;
+    config::enable_async_file_cache_write = false;
+    create_cache("read_ahead_requires_async_cache_writes");
+    ASSERT_FALSE(cache()->async_write_manager()->accepting());
+    ASSERT_FALSE(segment_v2::SegmentReadAhead::enabled());
+
+    auto source = std::make_shared<CountingFileReader>(open_remote_file());
+    auto reader = create_reader(source);
+    OlapReaderStatistics statistics;
+    StorageReadOptions read_options(statistics);
+    read_options.io_ctx.reader_type = ReaderType::READER_QUERY;
+    std::unique_ptr<segment_v2::SegmentReadAhead> read_ahead;
+    ASSERT_TRUE(segment_v2::SegmentReadAhead::create_for_query(reader, ExecEnv::GetInstance(),
+                                                               nullptr, read_options, &read_ahead)
+                        .ok());
+    EXPECT_EQ(read_ahead, nullptr);
+
+    std::string data(16, '\0');
+    size_t bytes_read = 0;
+    ASSERT_TRUE(reader->read_at(1024, Slice(data), &bytes_read).ok());
+    EXPECT_EQ(data, std::string(16, '0'));
+    ASSERT_EQ(source->read_count(), 1);
+    ASSERT_TRUE(reader->read_at(1024, Slice(data), &bytes_read).ok());
+    EXPECT_EQ(source->read_count(), 1);
+
+    config::enable_async_file_cache_write = true;
+    EXPECT_TRUE(segment_v2::SegmentReadAhead::enabled());
+    config::enable_query_read_ahead = false;
+    EXPECT_FALSE(segment_v2::SegmentReadAhead::enabled());
+}
+
+TEST_F(AsyncCachedRemoteFileReaderTest, cached_read_ahead_does_not_submit_writeback) {
+    create_cache("cached_read_ahead_no_writeback");
+    auto counting_reader = std::make_shared<CountingFileReader>(open_remote_file());
+    auto reader = create_reader(counting_reader);
+    std::string data(16, '\0');
+    size_t bytes_read = 0;
+    ASSERT_TRUE(reader->read_at(1024, Slice(data), &bytes_read).ok());
+    wait_for_async_writes();
+    const auto remote_reads = counting_reader->read_count();
+
+    auto* env = ExecEnv::GetInstance();
+    const auto old_deploy_mode = config::deploy_mode;
+    const bool old_read_ahead = config::enable_query_read_ahead;
+    auto old_scheduler = std::move(env->_file_range_read_scheduler);
+    auto old_partial_manager = std::move(env->_partial_block_writeback_manager);
+    std::unique_ptr<ThreadPool> executor;
+    Defer restore {[&]() {
+        env->_file_range_read_scheduler = std::move(old_scheduler);
+        env->_partial_block_writeback_manager = std::move(old_partial_manager);
+        config::deploy_mode = old_deploy_mode;
+        config::enable_query_read_ahead = old_read_ahead;
+    }};
+    config::deploy_mode = "cloud";
+    config::enable_query_read_ahead = true;
+    ASSERT_TRUE(ThreadPoolBuilder("CachedReadAheadTest").set_max_threads(1).build(&executor).ok());
+    ASSERT_TRUE(
+            FileRangeReadScheduler::create({.max_bytes_per_query = 1_mb, .max_bytes_per_be = 1_mb},
+                                           executor.get(), &env->_file_range_read_scheduler)
+                    .ok());
+    ASSERT_TRUE(PartialBlockWritebackManager::create(
+                        {.block_size = 1_mb,
+                         .worker_count = 1,
+                         .max_pending_bytes = 2_mb,
+                         .hole_fill_coalesce = {.max_gap_bytes = 32_kb,
+                                                .max_range_bytes = 1_mb,
+                                                .max_read_amplification_ratio = 1.0}},
+                        &env->_partial_block_writeback_manager)
+                        .ok());
+
+    OlapReaderStatistics statistics;
+    FileCacheStatistics cache_statistics;
+    StorageReadOptions read_options(statistics);
+    read_options.use_page_cache = false;
+    read_options.io_ctx.reader_type = ReaderType::READER_QUERY;
+    read_options.io_ctx.file_cache_stats = &cache_statistics;
+    std::unique_ptr<segment_v2::SegmentReadAhead> read_ahead;
+    ASSERT_TRUE(segment_v2::SegmentReadAhead::create_for_query(
+                        reader, env, env->file_range_read_scheduler()->create_context(),
+                        read_options, &read_ahead)
+                        .ok());
+    ASSERT_NE(read_ahead, nullptr);
+    std::unique_ptr<segment_v2::ColumnReadAhead> column;
+    ASSERT_TRUE(
+            segment_v2::ColumnReadAhead::create({{.page_index = 0,
+                                                  .first_ordinal = 0,
+                                                  .last_ordinal = 99,
+                                                  .range = {.offset = 1024, .size = data.size()}}},
+                                                {.window_bytes = data.size()}, false, &column)
+                    .ok());
+    const segment_v2::rowid_t rowid = 0;
+    roaring::Roaring rows;
+    rows.addRange(0, 100);
+    segment_v2::ColumnReadAheadPlan plan;
+    column->plan(&rowid, 1, rows, &plan);
+    ASSERT_TRUE(read_ahead->apply_plans({std::move(plan)}).accepted());
+    ASSERT_TRUE(read_ahead->file_reader()->read_at(1024, Slice(data), &bytes_read).ok());
+
+    EXPECT_EQ(cache_statistics.bytes_read_from_remote, 0);
+    EXPECT_EQ(cache_statistics.bytes_read_from_local, data.size());
+    EXPECT_EQ(counting_reader->read_count(), remote_reads);
+    EXPECT_EQ(statistics.read_ahead_stats->partial_blocks_queued.value(), 0);
+    EXPECT_EQ(statistics.read_ahead_stats->complete_blocks_submitted.value(), 0);
+    EXPECT_EQ(statistics.read_ahead_stats->writeback_deduplicated_blocks.value(), 0);
+    EXPECT_EQ(env->partial_block_writeback_manager()->pending_count(), 0);
+}
+
+TEST_F(AsyncCachedRemoteFileReaderTest, skips_consumed_range_writeback_for_no_write_reads) {
+    create_cache("cached_remote_reader_no_range_writeback");
+    auto reader = create_reader(open_remote_file());
+    std::unique_ptr<PartialBlockWritebackManager> partial_manager;
+    ASSERT_TRUE(PartialBlockWritebackManager::create(
+                        {.block_size = 1_mb,
+                         .worker_count = 1,
+                         .max_pending_bytes = 2_mb,
+                         .hole_fill_coalesce = {.max_gap_bytes = 32_kb,
+                                                .max_range_bytes = 1_mb,
+                                                .max_read_amplification_ratio = 2.0}},
+                        &partial_manager)
+                        .ok());
+
+    IOContext io_context;
+    io_context.file_cache_miss_policy = FileCacheMissPolicy::REMOTE_ONLY_ON_MISS;
+    EXPECT_EQ(reader->make_range_cache_writeback(io_context, partial_manager.get()), nullptr);
+
+    io_context.file_cache_miss_policy = FileCacheMissPolicy::READ_THROUGH_AND_WRITE_BACK;
+    io_context.cache_write_mode_override = CacheWriteMode::NO_WRITE;
+    EXPECT_EQ(reader->make_range_cache_writeback(io_context, partial_manager.get()), nullptr);
+}
+
 TEST_F(AsyncCachedRemoteFileReaderTest,
        downloading_preallocated_tail_can_finalize_short_while_reader_waits) {
     create_cache("cached_remote_reader_async_downloading_preallocated_tail");

be/test/storage/segment/column_reader_test.cpp

@@ -1051,8 +1051,8 @@ TEST_F(ColumnReaderTest, FileColumnIteratorFallsBackAfterReadAheadChecksumFailur
                                    .block_fill_min_coverage = 1.0},
                     .page_cache_probe = {},
                     .range_consumer_factory = [&]() {
-                        return [&](const io::FileRange& range, Slice) {
-                            consumed_ranges.push_back(range);
+                        return [&](const io::FileRangeRead& read) {
+                            consumed_ranges.push_back(read.range());
                         };
                     }});
 

be/test/storage/segment/segment_read_ahead_test.cpp

@@ -31,6 +31,7 @@
 #include "common/cast_set.h"
 #include "io/fs/read_ahead_metrics.h"
 #include "storage/olap_common.h"
+#include "storage/segment/column_reader.h"
 #include "storage/segment/page_handle.h"
 #include "storage/segment/page_io.h"
 #include "util/coding.h"
@@ -150,12 +151,14 @@ std::unique_ptr<io::FileRangeReadScheduler> make_scheduler(size_t max_bytes_per_
     return scheduler;
 }
 
-std::unique_ptr<SegmentReadAhead> make_segment_read_ahead(const io::FileReaderSPtr& source,
-                                                          io::FileRangeReadScheduler* scheduler,
-                                                          SegmentReadAheadOptions options) {
+std::unique_ptr<SegmentReadAhead> make_segment_read_ahead(
+        const io::FileReaderSPtr& source, io::FileRangeReadScheduler* scheduler,
+        SegmentReadAheadOptions options, ColumnReadAheadOptions eager_options = {},
+        ColumnReadAheadOptions lazy_options = {}) {
     auto context = scheduler->create_context();
     return std::make_unique<SegmentReadAhead>(source, scheduler, std::move(context),
-                                              io::FileRangeReadIOContext {}, std::move(options));
+                                              io::FileRangeReadIOContext {}, std::move(options),
+                                              eager_options, lazy_options);
 }
 
 roaring::Roaring rows(size_t count) {
@@ -170,12 +173,14 @@ TEST(SegmentReadAheadTest, CoalescesColumnsAndServesExactPageSlices) {
     auto scheduler = make_scheduler();
     size_t consumer_factory_calls = 0;
     std::vector<io::FileRange> consumed_ranges;
+    io::FileRangeReadStats consumed_stats;
     auto read_ahead = make_segment_read_ahead(
             source, scheduler.get(),
             {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = [&]() {
                  ++consumer_factory_calls;
-                 return [&](const io::FileRange& range, Slice) {
-                     consumed_ranges.push_back(range);
+                 return [&](const io::FileRangeRead& read) {
+                     consumed_ranges.push_back(read.range());
+                     consumed_stats = read.stats();
                  };
              }});
     auto first = make_window(1, 16, 0);
@@ -198,6 +203,10 @@ TEST(SegmentReadAheadTest, CoalescesColumnsAndServesExactPageSlices) {
     EXPECT_EQ(bytes_read, sizeof(first_data));
     ASSERT_EQ(consumed_ranges.size(), 1);
     EXPECT_EQ(consumed_ranges[0], (io::FileRange {.offset = 0, .size = 40}));
+    EXPECT_EQ(consumed_stats.file_cache.num_remote_io_total, 1);
+    EXPECT_EQ(consumed_stats.file_cache.bytes_read_from_remote, 40);
+    EXPECT_EQ(consumed_stats.file_reader.read_calls, 1);
+    EXPECT_EQ(consumed_stats.file_reader.read_bytes, 40);
 
     char second_data[16] {};
     ASSERT_TRUE(read_ahead->file_reader()
@@ -549,7 +558,7 @@ TEST(SegmentReadAheadTest, DiscardedUnusedRangeIsNotPublished) {
     auto read_ahead = make_segment_read_ahead(
             source, scheduler.get(),
             {.range_plan = plan_options(), .page_cache_probe = {}, .range_consumer_factory = [&]() {
-                 return [&](const io::FileRange&, Slice) { ++published; };
+                 return [&](const io::FileRangeRead&) { ++published; };
              }});
     auto window = make_window(2, 16);
     const auto scan_rows = rows(200);