diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index c1430c2ecfe5d2..8a104f35415521 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -47,6 +47,7 @@
 #include "common/exception.h"
 #include "common/signal_handler.h"
 #include "exec/tablet_info.h" // DorisNodesInfo
+#include "io/io_common.h"
 #include "olap/olap_common.h"
 #include "olap/rowset/beta_rowset.h"
 #include "olap/storage_engine.h"
@@ -82,6 +83,23 @@ namespace doris {
 
 #include "common/compile_check_begin.h"
 
+namespace {
+
+void set_topn_lazy_materialization_file_cache_stats(
+        const io::FileCacheStatistics& stats, PTopNLazyMaterializationFileCacheStats* pstats) {
+    pstats->set_local_io_count(stats.num_local_io_total);
+    pstats->set_local_io_bytes(stats.bytes_read_from_local);
+    pstats->set_remote_io_count(stats.num_remote_io_total);
+    pstats->set_remote_io_bytes(stats.bytes_read_from_remote);
+    pstats->set_skip_cache_io_count(stats.num_skip_cache_io_total);
+    pstats->set_write_cache_bytes(stats.bytes_write_into_cache);
+    pstats->set_local_io_time(stats.local_io_timer);
+    pstats->set_remote_io_time(stats.remote_io_timer);
+    pstats->set_write_cache_io_time(stats.write_cache_io_timer);
+}
+
+} // namespace
+
 Status RowIDFetcher::init() {
     DorisNodesInfo nodes_info;
     nodes_info.setNodes(_fetch_option.t_fetch_opt.nodes_info);
@@ -540,6 +558,11 @@ Status RowIdStorageReader::read_by_rowids(const PMultiGetRequestV2& request,
         int64_t external_get_block_avg_ms = 0;
         size_t external_scan_range_cnt = 0;
 
+        const auto file_cache_miss_policy =
+                request.file_cache_remote_only_on_miss()
+                        ? io::FileCacheMissPolicy::REMOTE_ONLY_ON_MISS
+                        : io::FileCacheMissPolicy::READ_THROUGH_AND_WRITE_BACK;
+
         // Add counters for different file mapping types
         std::unordered_map<FileMappingType, int64_t> file_type_counts;
 
@@ -590,7 +613,7 @@ Status RowIdStorageReader::read_by_rowids(const PMultiGetRequestV2& request,
                         RETURN_IF_ERROR(read_batch_doris_format_row(
                                 request_block_desc, id_file_map, slots, tquery_id, result_blocks[i],
                                 stats, &acquire_tablet_ms, &acquire_rowsets_ms,
-                                &acquire_segments_ms, &lookup_row_data_ms));
+                                &acquire_segments_ms, &lookup_row_data_ms, file_cache_miss_policy));
                     } else {
                         RETURN_IF_ERROR(read_batch_external_row(
                                 request.wg_id(), request_block_desc, id_file_map, slots,
@@ -641,6 +664,9 @@ Status RowIdStorageReader::read_by_rowids(const PMultiGetRequestV2& request,
                              acquire_rowsets_ms, acquire_segments_ms, lookup_row_data_ms,
                              file_type_stats, external_init_reader_avg_ms,
                              external_get_block_avg_ms, external_scan_range_cnt);
+        set_topn_lazy_materialization_file_cache_stats(
+                stats.file_cache_stats,
+                response->mutable_topn_lazy_materialization_file_cache_stats());
     }
 
     if (request.has_gc_id_map() && request.gc_id_map()) {
@@ -654,7 +680,8 @@ Status RowIdStorageReader::read_batch_doris_format_row(
         const PRequestBlockDesc& request_block_desc, std::shared_ptr<IdFileMap> id_file_map,
         std::vector<SlotDescriptor>& slots, const TUniqueId& query_id,
         vectorized::Block& result_block, OlapReaderStatistics& stats, int64_t* acquire_tablet_ms,
-        int64_t* acquire_rowsets_ms, int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms) {
+        int64_t* acquire_rowsets_ms, int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms,
+        io::FileCacheMissPolicy file_cache_miss_policy) {
     if (result_block.is_empty_column()) [[likely]] {
         result_block = vectorized::Block(slots, request_block_desc.row_id_size());
     }
@@ -698,7 +725,7 @@ Status RowIdStorageReader::read_batch_doris_format_row(
         RETURN_IF_ERROR(read_doris_format_row(
                 id_file_map, file_mapping, row_ids, slots, full_read_schema, row_store_read_struct,
                 stats, acquire_tablet_ms, acquire_rowsets_ms, acquire_segments_ms,
-                lookup_row_data_ms, seg_map, iterator_map, result_block));
+                lookup_row_data_ms, seg_map, iterator_map, file_cache_miss_policy, result_block));
 
         j += k;
         max_k = std::max(max_k, k);
@@ -712,6 +739,28 @@ const std::string RowIdStorageReader::ScannersRunningTimeProfile = "ScannersRunn
 const std::string RowIdStorageReader::InitReaderAvgTimeProfile = "InitReaderAvgTime";
 const std::string RowIdStorageReader::GetBlockAvgTimeProfile = "GetBlockAvgTime";
 const std::string RowIdStorageReader::FileReadLinesProfile = "FileReadLines";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOCount =
+        "TopNLazyMaterializationSecondPhaseLocalIOCount";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOBytes =
+        "TopNLazyMaterializationSecondPhaseLocalIOBytes";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOCount =
+        "TopNLazyMaterializationSecondPhaseRemoteIOCount";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOBytes =
+        "TopNLazyMaterializationSecondPhaseRemoteIOBytes";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseSkipCacheIOCount =
+        "TopNLazyMaterializationSecondPhaseSkipCacheIOCount";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheBytes =
+        "TopNLazyMaterializationSecondPhaseWriteCacheBytes";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOTime =
+        "TopNLazyMaterializationSecondPhaseLocalIOTime";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOTime =
+        "TopNLazyMaterializationSecondPhaseRemoteIOTime";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheIOTime =
+        "TopNLazyMaterializationSecondPhaseWriteCacheIOTime";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRowsRead =
+        "TopNLazyMaterializationSecondPhaseRowsRead";
+const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseSegmentsRead =
+        "TopNLazyMaterializationSecondPhaseSegmentsRead";
 
 Status RowIdStorageReader::read_batch_external_row(
         const uint64_t workload_group_id, const PRequestBlockDesc& request_block_desc,
@@ -1034,7 +1083,7 @@ Status RowIdStorageReader::read_doris_format_row(
         int64_t* acquire_tablet_ms, int64_t* acquire_rowsets_ms, int64_t* acquire_segments_ms,
         int64_t* lookup_row_data_ms, std::unordered_map<SegKey, SegItem, HashOfSegKey>& seg_map,
         std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey>& iterator_map,
-        vectorized::Block& result_block) {
+        io::FileCacheMissPolicy file_cache_miss_policy, vectorized::Block& result_block) {
     auto [tablet_id, rowset_id, segment_id] = file_mapping->get_doris_format_info();
     SegKey seg_key {.tablet_id = tablet_id, .rowset_id = rowset_id, .segment_id = segment_id};
 
@@ -1102,13 +1151,18 @@ Status RowIdStorageReader::read_doris_format_row(
             return Status::InternalError("Tablet {} does not have row store for all columns",
                                          tablet->tablet_id());
         }
+        io::IOContext io_ctx;
+        io_ctx.reader_type = ReaderType::READER_QUERY;
+        io_ctx.file_cache_stats = &stats.file_cache_stats;
+        io_ctx.file_cache_miss_policy = file_cache_miss_policy;
         for (auto row_id : row_ids) {
             RowLocation loc(rowset_id, segment->id(), cast_set<uint32_t>(row_id));
             row_store_read_struct.row_store_buffer.clear();
             RETURN_IF_ERROR(scope_timer_run(
                     [&]() {
                         return tablet->lookup_row_data({}, loc, rowset, stats,
-                                                       row_store_read_struct.row_store_buffer);
+                                                       row_store_read_struct.row_store_buffer,
+                                                       false, &io_ctx);
                     },
                     lookup_row_data_ms));
 
@@ -1131,6 +1185,8 @@ Status RowIdStorageReader::read_doris_format_row(
                 iterator_map[iterator_key].segment = segment;
                 iterator_item.storage_read_options.stats = &stats;
                 iterator_item.storage_read_options.io_ctx.reader_type = ReaderType::READER_QUERY;
+                iterator_item.storage_read_options.io_ctx.file_cache_miss_policy =
+                        file_cache_miss_policy;
             }
             for (auto row_id : row_ids) {
                 RETURN_IF_ERROR(segment->seek_and_read_by_rowid(
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index c617678ccd8e2d..6164011bc61c8b 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -38,6 +38,9 @@ namespace doris {
 class DorisNodesInfo;
 class RuntimeState;
 class TupleDescriptor;
+namespace io {
+enum class FileCacheMissPolicy : uint8_t;
+}
 
 struct FileMapping;
 struct SegKey;
@@ -100,6 +103,17 @@ class RowIdStorageReader {
     static const std::string InitReaderAvgTimeProfile;
     static const std::string GetBlockAvgTimeProfile;
     static const std::string FileReadLinesProfile;
+    static const std::string TopNLazyMaterializationSecondPhaseLocalIOCount;
+    static const std::string TopNLazyMaterializationSecondPhaseLocalIOBytes;
+    static const std::string TopNLazyMaterializationSecondPhaseRemoteIOCount;
+    static const std::string TopNLazyMaterializationSecondPhaseRemoteIOBytes;
+    static const std::string TopNLazyMaterializationSecondPhaseSkipCacheIOCount;
+    static const std::string TopNLazyMaterializationSecondPhaseWriteCacheBytes;
+    static const std::string TopNLazyMaterializationSecondPhaseLocalIOTime;
+    static const std::string TopNLazyMaterializationSecondPhaseRemoteIOTime;
+    static const std::string TopNLazyMaterializationSecondPhaseWriteCacheIOTime;
+    static const std::string TopNLazyMaterializationSecondPhaseRowsRead;
+    static const std::string TopNLazyMaterializationSecondPhaseSegmentsRead;
 
     static Status read_by_rowids(const PMultiGetRequest& request, PMultiGetResponse* response);
     static Status read_by_rowids(const PMultiGetRequestV2& request, PMultiGetResponseV2* response);
@@ -113,14 +127,14 @@ class RowIdStorageReader {
             int64_t* acquire_tablet_ms, int64_t* acquire_rowsets_ms, int64_t* acquire_segments_ms,
             int64_t* lookup_row_data_ms, std::unordered_map<SegKey, SegItem, HashOfSegKey>& seg_map,
             std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey>& iterator_map,
-            vectorized::Block& result_block);
+            io::FileCacheMissPolicy file_cache_miss_policy, vectorized::Block& result_block);
 
     static Status read_batch_doris_format_row(
             const PRequestBlockDesc& request_block_desc, std::shared_ptr<IdFileMap> id_file_map,
             std::vector<SlotDescriptor>& slots, const TUniqueId& query_id,
             vectorized::Block& result_block, OlapReaderStatistics& stats,
             int64_t* acquire_tablet_ms, int64_t* acquire_rowsets_ms, int64_t* acquire_segments_ms,
-            int64_t* lookup_row_data_ms);
+            int64_t* lookup_row_data_ms, io::FileCacheMissPolicy file_cache_miss_policy);
 
     static Status read_batch_external_row(
             const uint64_t workload_group_id, const PRequestBlockDesc& request_block_desc,
diff --git a/be/src/io/cache/block_file_cache.cpp b/be/src/io/cache/block_file_cache.cpp
index 282d6a1dac1943..7fd5177e44b57f 100644
--- a/be/src/io/cache/block_file_cache.cpp
+++ b/be/src/io/cache/block_file_cache.cpp
@@ -785,6 +785,96 @@ void BlockFileCache::add_need_update_lru_block(FileBlockSPtr block) {
     }
 }
 
+Status BlockFileCache::get_downloaded_blocks_if_fully_covered(const UInt128Wrapper& hash,
+                                                              size_t offset, size_t size,
+                                                              const CacheContext& context,
+                                                              FileBlocks* blocks,
+                                                              bool* fully_covered) {
+    DCHECK(blocks != nullptr);
+    DCHECK(fully_covered != nullptr);
+    blocks->clear();
+    *fully_covered = false;
+    if (size == 0) {
+        *fully_covered = true;
+        return Status::OK();
+    }
+
+    FileBlock::Range range(offset, offset + size - 1);
+    std::lock_guard cache_lock(_mutex);
+    auto it = _files.find(hash);
+    if (it == _files.end()) {
+        if (_async_open_done) {
+            return Status::OK();
+        }
+        FileCacheKey key;
+        key.hash = hash;
+        key.meta.type = context.cache_type;
+        key.meta.expiration_time = context.expiration_time;
+        _storage->load_blocks_directly_unlocked(this, key, cache_lock);
+
+        it = _files.find(hash);
+        if (it == _files.end()) {
+            return Status::OK();
+        }
+    }
+
+    auto& file_blocks = it->second;
+    if (file_blocks.empty()) {
+        LOG(WARNING) << "file_blocks is empty for hash=" << hash.to_string()
+                     << " cache type=" << context.cache_type
+                     << " cache expiration time=" << context.expiration_time
+                     << " cache range=" << range.left << " " << range.right
+                     << " query id=" << context.query_id;
+        DCHECK(false);
+        _files.erase(hash);
+        return Status::OK();
+    }
+
+    std::vector<FileBlockCell*> covered_cells;
+    auto block_it = file_blocks.lower_bound(range.left);
+    if (block_it == file_blocks.end() || block_it->second.file_block->range().left > range.left) {
+        if (block_it == file_blocks.begin()) {
+            return Status::OK();
+        }
+        --block_it;
+    }
+
+    size_t current_pos = range.left;
+    while (current_pos <= range.right) {
+        if (block_it == file_blocks.end()) {
+            return Status::OK();
+        }
+
+        auto& cell = block_it->second;
+        const auto& block_range = cell.file_block->range();
+        if (block_range.right < current_pos) {
+            ++block_it;
+            continue;
+        }
+        if (block_range.left > current_pos ||
+            cell.file_block->state() != FileBlock::State::DOWNLOADED) {
+            return Status::OK();
+        }
+
+        covered_cells.push_back(&cell);
+        if (range.right <= block_range.right) {
+            *fully_covered = true;
+            break;
+        }
+        current_pos = block_range.right + 1;
+        ++block_it;
+    }
+
+    if (!*fully_covered) {
+        return Status::OK();
+    }
+    for (const auto* cell : covered_cells) {
+        use_cell(*cell, blocks, need_to_move(cell->file_block->cache_type(), context.cache_type),
+                 cache_lock);
+    }
+    return Status::OK();
+}
+
 std::string BlockFileCache::clear_file_cache_async() {
     return clear_file_cache_impl(false);
 }
diff --git a/be/src/io/cache/block_file_cache.h b/be/src/io/cache/block_file_cache.h
index fb610d3c7505d7..598e08bb51f49d 100644
--- a/be/src/io/cache/block_file_cache.h
+++ b/be/src/io/cache/block_file_cache.h
@@ -230,6 +230,14 @@ class BlockFileCache {
     FileBlocksHolder get_or_set(const UInt128Wrapper& hash, size_t offset, size_t size,
                                 CacheContext& context);
 
+    /**
+     * Return existing downloaded blocks only if they fully cover [offset, offset + size).
+     * This lookup is read-only: it does not reserve cache space or create EMPTY blocks.
+     */
+    Status get_downloaded_blocks_if_fully_covered(const UInt128Wrapper& hash, size_t offset,
+                                                  size_t size, const CacheContext& context,
+                                                  FileBlocks* blocks, bool* fully_covered);
+
     /**
      * record blocks read directly by CachedRemoteFileReader
      */
diff --git a/be/src/io/cache/cached_remote_file_reader.cpp b/be/src/io/cache/cached_remote_file_reader.cpp
index b4f019210ed035..b971b55c34bbd5 100644
--- a/be/src/io/cache/cached_remote_file_reader.cpp
+++ b/be/src/io/cache/cached_remote_file_reader.cpp
@@ -273,6 +273,96 @@ Status CachedRemoteFileReader::_execute_remote_read(const std::vector<FileBlockS
     }
 }
 
+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,
+                                                               const IOContext* io_ctx) {
+    auto read_remote = [&]() -> Status {
+        stats.hit_cache = false;
+        stats.from_peer_cache = false;
+        stats.skip_cache = true;
+        s3_read_counter << 1;
+        if (is_dryrun) [[unlikely]] {
+            *bytes_read = bytes_req;
+            g_read_cache_indirect_bytes << 0;
+            g_read_cache_indirect_total_bytes << bytes_req;
+            return Status::OK();
+        }
+
+        size_t remote_bytes_read = bytes_req;
+        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));
+        *bytes_read = remote_bytes_read;
+        DCHECK_EQ(*bytes_read, bytes_req);
+        stats.bytes_read_from_remote += remote_bytes_read;
+        g_read_cache_indirect_bytes << remote_bytes_read;
+        g_read_cache_indirect_total_bytes << remote_bytes_read;
+        return Status::OK();
+    };
+
+    g_read_cache_indirect_num << 1;
+    CacheContext cache_context(io_ctx);
+    cache_context.stats = &stats;
+    cache_context.tablet_id = _tablet_id;
+    FileBlocks file_blocks;
+    bool fully_covered = false;
+    {
+        SCOPED_RAW_TIMER(&stats.get_timer);
+        RETURN_IF_ERROR(_cache->get_downloaded_blocks_if_fully_covered(
+                _cache_hash, offset, bytes_req, cache_context, &file_blocks, &fully_covered));
+    }
+    if (!fully_covered) {
+        return read_remote();
+    }
+
+    size_t local_read_bytes = 0;
+    size_t current_offset = offset;
+    size_t end_offset = offset + bytes_req - 1;
+    for (auto& block : file_blocks) {
+        if (current_offset > end_offset) {
+            break;
+        }
+        const auto& block_range = block->range();
+        if (block_range.right < current_offset) {
+            continue;
+        }
+
+        size_t read_left = std::max(current_offset, block_range.left);
+        size_t read_right = std::min(end_offset, block_range.right);
+        size_t read_size = read_right - read_left + 1;
+        if (is_dryrun) [[unlikely]] {
+            g_skip_local_cache_io_sum_bytes << read_size;
+        } else {
+            SCOPED_RAW_TIMER(&stats.local_read_timer);
+            SCOPED_CONCURRENCY_COUNT(
+                    ConcurrencyStatsManager::instance().cached_remote_reader_local_read);
+            Status st = block->read(Slice(result.data + (read_left - offset), read_size),
+                                    read_left - block_range.left);
+            if (!st.ok()) {
+                if (st.is<ErrorCode::NOT_FOUND>()) {
+                    _cache->remove_if_cached_async(_cache_hash);
+                }
+                LOG_EVERY_N(WARNING, 100)
+                        << "Read data failed from file cache in remote-only-on-miss path. "
+                        << "Fallback to remote. err=" << st.msg()
+                        << ", block state=" << block->state();
+                return read_remote();
+            }
+            stats.bytes_read_from_local += read_size;
+            local_read_bytes += read_size;
+        }
+        current_offset = read_right + 1;
+    }
+
+    *bytes_read = bytes_req;
+    stats.hit_cache = true;
+    g_read_cache_indirect_bytes << local_read_bytes;
+    g_read_cache_indirect_total_bytes << bytes_req;
+    return Status::OK();
+}
+
 Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                                             const IOContext* io_ctx) {
     size_t already_read = 0;
@@ -339,6 +429,12 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t*
         }
     };
     std::unique_ptr<int, decltype(defer_func)> defer((int*)0x01, std::move(defer_func));
+    if (io_ctx->file_cache_miss_policy == FileCacheMissPolicy::REMOTE_ONLY_ON_MISS) {
+        Status st = _read_remote_only_on_cache_miss(offset, result, bytes_req, is_dryrun,
+                                                    bytes_read, stats, io_ctx);
+        read_success = st.ok();
+        return st;
+    }
     if (_is_doris_table && config::enable_read_cache_file_directly) {
         // read directly
         SCOPED_RAW_TIMER(&stats.read_cache_file_directly_timer);
diff --git a/be/src/io/cache/cached_remote_file_reader.h b/be/src/io/cache/cached_remote_file_reader.h
index ddf75e246aa29f..e6a07e5071fe96 100644
--- a/be/src/io/cache/cached_remote_file_reader.h
+++ b/be/src/io/cache/cached_remote_file_reader.h
@@ -86,6 +86,12 @@ class CachedRemoteFileReader final : public FileReader,
                                 size_t& size, std::unique_ptr<char[]>& buffer,
                                 ReadStatistics& stats, const IOContext* io_ctx);
 
+    /// Read local cache only when downloaded blocks fully cover the request; otherwise read remote
+    /// data directly without writing file cache.
+    Status _read_remote_only_on_cache_miss(size_t offset, Slice result, size_t bytes_req,
+                                           bool is_dryrun, size_t* bytes_read,
+                                           ReadStatistics& stats, const IOContext* io_ctx);
+
     void _update_stats(const ReadStatistics& stats, FileCacheStatistics* state,
                        FileCacheReadType read_type) const;
 
diff --git a/be/src/io/io_common.h b/be/src/io/io_common.h
index 031387708ea6c5..6a350d1bdde7d4 100644
--- a/be/src/io/io_common.h
+++ b/be/src/io/io_common.h
@@ -35,6 +35,11 @@ enum class ReaderType : uint8_t {
 
 namespace io {
 
+enum class FileCacheMissPolicy : uint8_t {
+    READ_THROUGH_AND_WRITE_BACK = 0,
+    REMOTE_ONLY_ON_MISS = 1,
+};
+
 struct FileReaderStats {
     size_t read_calls = 0;
     size_t read_bytes = 0;
@@ -104,6 +109,7 @@ struct IOContext {
     bool is_dryrun = false;
     // if `is_warmup` == true, this I/O request is from a warm up task
     bool is_warmup {false};
+    FileCacheMissPolicy file_cache_miss_policy = FileCacheMissPolicy::READ_THROUGH_AND_WRITE_BACK;
 };
 
 } // namespace io
diff --git a/be/src/olap/base_tablet.cpp b/be/src/olap/base_tablet.cpp
index 33bf5dcc132c0d..6a6ed469466d22 100644
--- a/be/src/olap/base_tablet.cpp
+++ b/be/src/olap/base_tablet.cpp
@@ -78,7 +78,8 @@ Status _get_segment_column_iterator(const BetaRowsetSharedPtr& rowset, uint32_t
                                     const TabletColumn& target_column,
                                     SegmentCacheHandle* segment_cache_handle,
                                     std::unique_ptr<segment_v2::ColumnIterator>* column_iterator,
-                                    OlapReaderStatistics* stats) {
+                                    OlapReaderStatistics* stats,
+                                    const io::IOContext* input_io_ctx = nullptr) {
     RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(rowset, segment_cache_handle, true));
     // find segment
     auto it = std::find_if(
@@ -92,13 +93,18 @@ Status _get_segment_column_iterator(const BetaRowsetSharedPtr& rowset, uint32_t
     segment_v2::SegmentSharedPtr segment = *it;
     StorageReadOptions opts;
     opts.stats = stats;
+    if (input_io_ctx != nullptr) {
+        opts.io_ctx = *input_io_ctx;
+    }
     RETURN_IF_ERROR(segment->new_column_iterator(target_column, column_iterator, &opts));
+    auto io_ctx = opts.io_ctx;
+    io_ctx.reader_type = ReaderType::READER_QUERY;
+    io_ctx.file_cache_stats = &stats->file_cache_stats;
     segment_v2::ColumnIteratorOptions opt {
             .use_page_cache = !config::disable_storage_page_cache,
             .file_reader = segment->file_reader().get(),
             .stats = stats,
-            .io_ctx = io::IOContext {.reader_type = ReaderType::READER_QUERY,
-                                     .file_cache_stats = &stats->file_cache_stats},
+            .io_ctx = io_ctx,
     };
     RETURN_IF_ERROR((*column_iterator)->init(opt));
     return Status::OK();
@@ -408,7 +414,8 @@ std::vector<RowsetSharedPtr> BaseTablet::get_rowset_by_ids(
 
 Status BaseTablet::lookup_row_data(const Slice& encoded_key, const RowLocation& row_location,
                                    RowsetSharedPtr input_rowset, OlapReaderStatistics& stats,
-                                   std::string& values, bool write_to_cache) {
+                                   std::string& values, bool write_to_cache,
+                                   const io::IOContext* io_ctx) {
     MonotonicStopWatch watch;
     size_t row_size = 1;
     watch.start();
@@ -424,7 +431,8 @@ Status BaseTablet::lookup_row_data(const Slice& encoded_key, const RowLocation&
     std::unique_ptr<segment_v2::ColumnIterator> column_iterator;
     const auto& column = *DORIS_TRY(tablet_schema->column(BeConsts::ROW_STORE_COL));
     RETURN_IF_ERROR(_get_segment_column_iterator(rowset, row_location.segment_id, column,
-                                                 &segment_cache_handle, &column_iterator, &stats));
+                                                 &segment_cache_handle, &column_iterator, &stats,
+                                                 io_ctx));
     // get and parse tuple row
     vectorized::MutableColumnPtr column_ptr = vectorized::ColumnString::create();
     std::vector<segment_v2::rowid_t> rowids {static_cast<segment_v2::rowid_t>(row_location.row_id)};
diff --git a/be/src/olap/base_tablet.h b/be/src/olap/base_tablet.h
index ff45ef4163a8f0..47fa8654961aa1 100644
--- a/be/src/olap/base_tablet.h
+++ b/be/src/olap/base_tablet.h
@@ -163,7 +163,7 @@ class BaseTablet : public std::enable_shared_from_this<BaseTablet> {
     // Lookup a row with TupleDescriptor and fill Block
     Status lookup_row_data(const Slice& encoded_key, const RowLocation& row_location,
                            RowsetSharedPtr rowset, OlapReaderStatistics& stats, std::string& values,
-                           bool write_to_cache = false);
+                           bool write_to_cache = false, const io::IOContext* io_ctx = nullptr);
     // Lookup the row location of `encoded_key`, the function sets `row_location` on success.
     // NOTE: the method only works in unique key model with primary key index, you will got a
     //       not supported error in other data model.
diff --git a/be/src/olap/rowset/segment_v2/bloom_filter_index_reader.cpp b/be/src/olap/rowset/segment_v2/bloom_filter_index_reader.cpp
index 3c638875a86921..8d77384cbd5bbd 100644
--- a/be/src/olap/rowset/segment_v2/bloom_filter_index_reader.cpp
+++ b/be/src/olap/rowset/segment_v2/bloom_filter_index_reader.cpp
@@ -33,10 +33,11 @@
 namespace doris::segment_v2 {
 #include "common/compile_check_begin.h"
 Status BloomFilterIndexReader::load(bool use_page_cache, bool kept_in_memory,
-                                    OlapReaderStatistics* index_load_stats) {
+                                    OlapReaderStatistics* index_load_stats,
+                                    const io::IOContext* io_ctx) {
     // TODO yyq: implement a new once flag to avoid status construct.
-    return _load_once.call([this, use_page_cache, kept_in_memory, index_load_stats] {
-        return _load(use_page_cache, kept_in_memory, index_load_stats);
+    return _load_once.call([this, use_page_cache, kept_in_memory, index_load_stats, io_ctx] {
+        return _load(use_page_cache, kept_in_memory, index_load_stats, io_ctx);
     });
 }
 
@@ -46,21 +47,24 @@ int64_t BloomFilterIndexReader::get_metadata_size() const {
 }
 
 Status BloomFilterIndexReader::_load(bool use_page_cache, bool kept_in_memory,
-                                     OlapReaderStatistics* index_load_stats) {
+                                     OlapReaderStatistics* index_load_stats,
+                                     const io::IOContext* io_ctx) {
     const IndexedColumnMetaPB& bf_index_meta = _bloom_filter_index_meta->bloom_filter();
 
     _bloom_filter_reader = std::make_unique<IndexedColumnReader>(_file_reader, bf_index_meta);
-    RETURN_IF_ERROR(_bloom_filter_reader->load(use_page_cache, kept_in_memory, index_load_stats));
+    RETURN_IF_ERROR(
+            _bloom_filter_reader->load(use_page_cache, kept_in_memory, index_load_stats, io_ctx));
     update_metadata_size();
     return Status::OK();
 }
 
 Status BloomFilterIndexReader::new_iterator(std::unique_ptr<BloomFilterIndexIterator>* iterator,
-                                            OlapReaderStatistics* index_load_stats) {
+                                            OlapReaderStatistics* index_load_stats,
+                                            const io::IOContext* io_ctx) {
     DBUG_EXECUTE_IF("BloomFilterIndexReader::new_iterator.fail", {
         return Status::InternalError("new_iterator for bloom filter index failed");
     });
-    *iterator = std::make_unique<BloomFilterIndexIterator>(this, index_load_stats);
+    *iterator = std::make_unique<BloomFilterIndexIterator>(this, index_load_stats, io_ctx);
     return Status::OK();
 }
 
diff --git a/be/src/olap/rowset/segment_v2/bloom_filter_index_reader.h b/be/src/olap/rowset/segment_v2/bloom_filter_index_reader.h
index fb53af89c0fe92..22c3dceff5149d 100644
--- a/be/src/olap/rowset/segment_v2/bloom_filter_index_reader.h
+++ b/be/src/olap/rowset/segment_v2/bloom_filter_index_reader.h
@@ -31,6 +31,9 @@
 #include "util/once.h"
 
 namespace doris {
+namespace io {
+struct IOContext;
+}
 
 namespace segment_v2 {
 
@@ -47,19 +50,21 @@ class BloomFilterIndexReader : public MetadataAdder<BloomFilterIndexReader> {
         _bloom_filter_index_meta.reset(new BloomFilterIndexPB(bloom_filter_index_meta));
     }
 
-    Status load(bool use_page_cache, bool kept_in_memory,
-                OlapReaderStatistics* bf_index_load_stats);
+    Status load(bool use_page_cache, bool kept_in_memory, OlapReaderStatistics* bf_index_load_stats,
+                const io::IOContext* io_ctx = nullptr);
 
     BloomFilterAlgorithmPB algorithm() { return _bloom_filter_index_meta->algorithm(); }
 
     // create a new column iterator.
     Status new_iterator(std::unique_ptr<BloomFilterIndexIterator>* iterator,
-                        OlapReaderStatistics* index_load_stats);
+                        OlapReaderStatistics* index_load_stats,
+                        const io::IOContext* io_ctx = nullptr);
 
     const TypeInfo* type_info() const { return _type_info; }
 
 private:
-    Status _load(bool use_page_cache, bool kept_in_memory, OlapReaderStatistics* index_load_stats);
+    Status _load(bool use_page_cache, bool kept_in_memory, OlapReaderStatistics* index_load_stats,
+                 const io::IOContext* io_ctx);
 
     int64_t get_metadata_size() const override;
 
@@ -75,8 +80,10 @@ class BloomFilterIndexReader : public MetadataAdder<BloomFilterIndexReader> {
 
 class BloomFilterIndexIterator {
 public:
-    explicit BloomFilterIndexIterator(BloomFilterIndexReader* reader, OlapReaderStatistics* stats)
-            : _reader(reader), _bloom_filter_iter(reader->_bloom_filter_reader.get(), stats) {}
+    explicit BloomFilterIndexIterator(BloomFilterIndexReader* reader, OlapReaderStatistics* stats,
+                                      const io::IOContext* io_ctx)
+            : _reader(reader),
+              _bloom_filter_iter(reader->_bloom_filter_reader.get(), stats, io_ctx) {}
 
     // Read bloom filter at the given ordinal into `bf`.
     Status read_bloom_filter(rowid_t ordinal, std::unique_ptr<BloomFilter>* bf);
diff --git a/be/src/olap/rowset/segment_v2/column_reader.cpp b/be/src/olap/rowset/segment_v2/column_reader.cpp
index 63313ac4458cca..b4df8e5da708f5 100644
--- a/be/src/olap/rowset/segment_v2/column_reader.cpp
+++ b/be/src/olap/rowset/segment_v2/column_reader.cpp
@@ -629,7 +629,8 @@ Status ColumnReader::get_row_ranges_by_bloom_filter(const AndBlockColumnPredicat
             _load_bloom_filter_index(_use_index_page_cache, _opts.kept_in_memory, iter_opts));
     RowRanges bf_row_ranges;
     std::unique_ptr<BloomFilterIndexIterator> bf_iter;
-    RETURN_IF_ERROR(_bloom_filter_index->new_iterator(&bf_iter, iter_opts.stats));
+    RETURN_IF_ERROR(
+            _bloom_filter_index->new_iterator(&bf_iter, iter_opts.stats, &iter_opts.io_ctx));
     size_t range_size = row_ranges->range_size();
     // get covered page ids
     std::set<uint32_t> page_ids;
@@ -661,13 +662,14 @@ Status ColumnReader::_load_ordinal_index(bool use_page_cache, bool kept_in_memor
     if (!_ordinal_index) {
         return Status::InternalError("ordinal_index not inited");
     }
-    return _ordinal_index->load(use_page_cache, kept_in_memory, iter_opts.stats);
+    return _ordinal_index->load(use_page_cache, kept_in_memory, iter_opts.stats, &iter_opts.io_ctx);
 }
 
 Status ColumnReader::_load_zone_map_index(bool use_page_cache, bool kept_in_memory,
                                           const ColumnIteratorOptions& iter_opts) {
     if (_zone_map_index != nullptr) {
-        return _zone_map_index->load(use_page_cache, kept_in_memory, iter_opts.stats);
+        return _zone_map_index->load(use_page_cache, kept_in_memory, iter_opts.stats,
+                                     &iter_opts.io_ctx);
     }
     return Status::OK();
 }
@@ -749,7 +751,8 @@ bool ColumnReader::has_bloom_filter_index(bool ngram) const {
 Status ColumnReader::_load_bloom_filter_index(bool use_page_cache, bool kept_in_memory,
                                               const ColumnIteratorOptions& iter_opts) {
     if (_bloom_filter_index != nullptr) {
-        return _bloom_filter_index->load(use_page_cache, kept_in_memory, iter_opts.stats);
+        return _bloom_filter_index->load(use_page_cache, kept_in_memory, iter_opts.stats,
+                                         &iter_opts.io_ctx);
     }
     return Status::OK();
 }
diff --git a/be/src/olap/rowset/segment_v2/indexed_column_reader.cpp b/be/src/olap/rowset/segment_v2/indexed_column_reader.cpp
index 62325f1dbe247f..edee7e6264ff2d 100644
--- a/be/src/olap/rowset/segment_v2/indexed_column_reader.cpp
+++ b/be/src/olap/rowset/segment_v2/indexed_column_reader.cpp
@@ -60,7 +60,8 @@ int64_t IndexedColumnReader::get_metadata_size() const {
 }
 
 Status IndexedColumnReader::load(bool use_page_cache, bool kept_in_memory,
-                                 OlapReaderStatistics* index_load_stats) {
+                                 OlapReaderStatistics* index_load_stats,
+                                 const io::IOContext* io_ctx) {
     _use_page_cache = use_page_cache;
     _kept_in_memory = kept_in_memory;
 
@@ -78,7 +79,7 @@ Status IndexedColumnReader::load(bool use_page_cache, bool kept_in_memory,
         } else {
             RETURN_IF_ERROR(load_index_page(_meta.ordinal_index_meta().root_page(),
                                             &_ordinal_index_page_handle,
-                                            _ordinal_index_reader.get(), index_load_stats));
+                                            _ordinal_index_reader.get(), index_load_stats, io_ctx));
             _has_index_page = true;
         }
     }
@@ -90,7 +91,7 @@ Status IndexedColumnReader::load(bool use_page_cache, bool kept_in_memory,
         } else {
             RETURN_IF_ERROR(load_index_page(_meta.value_index_meta().root_page(),
                                             &_value_index_page_handle, _value_index_reader.get(),
-                                            index_load_stats));
+                                            index_load_stats, io_ctx));
             _has_index_page = true;
         }
     }
@@ -102,13 +103,14 @@ Status IndexedColumnReader::load(bool use_page_cache, bool kept_in_memory,
 
 Status IndexedColumnReader::load_index_page(const PagePointerPB& pp, PageHandle* handle,
                                             IndexPageReader* reader,
-                                            OlapReaderStatistics* index_load_stats) {
+                                            OlapReaderStatistics* index_load_stats,
+                                            const io::IOContext* io_ctx) {
     Slice body;
     PageFooterPB footer;
     BlockCompressionCodec* local_compress_codec;
     RETURN_IF_ERROR(get_block_compression_codec(_meta.compression(), &local_compress_codec));
     RETURN_IF_ERROR(read_page(PagePointer(pp), handle, &body, &footer, INDEX_PAGE,
-                              local_compress_codec, false, index_load_stats));
+                              local_compress_codec, false, index_load_stats, io_ctx));
     RETURN_IF_ERROR(reader->parse(body, footer.index_page_footer()));
     _mem_size += body.get_size();
     return Status::OK();
@@ -117,11 +119,14 @@ Status IndexedColumnReader::load_index_page(const PagePointerPB& pp, PageHandle*
 Status IndexedColumnReader::read_page(const PagePointer& pp, PageHandle* handle, Slice* body,
                                       PageFooterPB* footer, PageTypePB type,
                                       BlockCompressionCodec* codec, bool pre_decode,
-                                      OlapReaderStatistics* stats) const {
+                                      OlapReaderStatistics* stats,
+                                      const io::IOContext* io_ctx) const {
     OlapReaderStatistics tmp_stats;
     OlapReaderStatistics* stats_ptr = stats != nullptr ? stats : &tmp_stats;
-    PageReadOptions opts(io::IOContext {.is_index_data = true,
-                                        .file_cache_stats = &stats_ptr->file_cache_stats});
+    io::IOContext page_io_ctx = io_ctx != nullptr ? *io_ctx : io::IOContext {};
+    page_io_ctx.is_index_data = true;
+    page_io_ctx.file_cache_stats = &stats_ptr->file_cache_stats;
+    PageReadOptions opts(page_io_ctx);
     opts.use_page_cache = _use_page_cache;
     opts.kept_in_memory = _kept_in_memory;
     opts.pre_decode = pre_decode;
@@ -158,7 +163,7 @@ Status IndexedColumnIterator::_read_data_page(const PagePointer& pp) {
     Slice body;
     PageFooterPB footer;
     RETURN_IF_ERROR(_reader->read_page(pp, &handle, &body, &footer, DATA_PAGE, _compress_codec,
-                                       true, _stats));
+                                       true, _stats, _io_ctx));
     // parse data page
     // note that page_index is not used in IndexedColumnIterator, so we pass 0
     PageDecoderOptions opts;
diff --git a/be/src/olap/rowset/segment_v2/indexed_column_reader.h b/be/src/olap/rowset/segment_v2/indexed_column_reader.h
index 6e62feaafdcdd1..4225f9f6b7e7e1 100644
--- a/be/src/olap/rowset/segment_v2/indexed_column_reader.h
+++ b/be/src/olap/rowset/segment_v2/indexed_column_reader.h
@@ -41,6 +41,9 @@ namespace doris {
 class KeyCoder;
 class TypeInfo;
 class BlockCompressionCodec;
+namespace io {
+struct IOContext;
+}
 
 namespace segment_v2 {
 
@@ -58,12 +61,14 @@ class IndexedColumnReader : public MetadataAdder<IndexedColumnReader> {
     ~IndexedColumnReader() override;
 
     Status load(bool use_page_cache, bool kept_in_memory,
-                OlapReaderStatistics* index_load_stats = nullptr);
+                OlapReaderStatistics* index_load_stats = nullptr,
+                const io::IOContext* io_ctx = nullptr);
 
     // read a page specified by `pp' from `file' into `handle'
     Status read_page(const PagePointer& pp, PageHandle* handle, Slice* body, PageFooterPB* footer,
                      PageTypePB type, BlockCompressionCodec* codec, bool pre_decode,
-                     OlapReaderStatistics* stats = nullptr) const;
+                     OlapReaderStatistics* stats = nullptr,
+                     const io::IOContext* io_ctx = nullptr) const;
 
     int64_t num_values() const { return _num_values; }
     const EncodingInfo* encoding_info() const { return _encoding_info; }
@@ -77,7 +82,7 @@ class IndexedColumnReader : public MetadataAdder<IndexedColumnReader> {
 
 private:
     Status load_index_page(const PagePointerPB& pp, PageHandle* handle, IndexPageReader* reader,
-                           OlapReaderStatistics* index_load_stats);
+                           OlapReaderStatistics* index_load_stats, const io::IOContext* io_ctx);
 
     int64_t get_metadata_size() const override;
 
@@ -109,11 +114,13 @@ class IndexedColumnReader : public MetadataAdder<IndexedColumnReader> {
 class IndexedColumnIterator {
 public:
     explicit IndexedColumnIterator(const IndexedColumnReader* reader,
-                                   OlapReaderStatistics* stats = nullptr)
+                                   OlapReaderStatistics* stats = nullptr,
+                                   const io::IOContext* io_ctx = nullptr)
             : _reader(reader),
               _ordinal_iter(reader->_ordinal_index_reader.get()),
               _value_iter(reader->_value_index_reader.get()),
-              _stats(stats) {}
+              _stats(stats),
+              _io_ctx(io_ctx) {}
 
     // Seek to the given ordinal entry. Entry 0 is the first entry.
     // Return Status::Error<ENTRY_NOT_FOUND> if provided seek point is past the end.
@@ -163,6 +170,7 @@ class IndexedColumnIterator {
     // iterator owned compress codec, should NOT be shared by threads, initialized before used
     BlockCompressionCodec* _compress_codec = nullptr;
     OlapReaderStatistics* _stats = nullptr;
+    const io::IOContext* _io_ctx = nullptr;
 };
 
 } // namespace segment_v2
diff --git a/be/src/olap/rowset/segment_v2/ordinal_page_index.cpp b/be/src/olap/rowset/segment_v2/ordinal_page_index.cpp
index 54310186f55eac..f35e7450296341 100644
--- a/be/src/olap/rowset/segment_v2/ordinal_page_index.cpp
+++ b/be/src/olap/rowset/segment_v2/ordinal_page_index.cpp
@@ -72,16 +72,17 @@ Status OrdinalIndexWriter::finish(io::FileWriter* file_writer, ColumnIndexMetaPB
 }
 
 Status OrdinalIndexReader::load(bool use_page_cache, bool kept_in_memory,
-                                OlapReaderStatistics* index_load_stats) {
+                                OlapReaderStatistics* index_load_stats,
+                                const io::IOContext* io_ctx) {
     // TODO yyq: implement a new once flag to avoid status construct.
-    return _load_once.call([this, use_page_cache, kept_in_memory, index_load_stats] {
-        return _load(use_page_cache, kept_in_memory, std::move(_meta_pb), index_load_stats);
+    return _load_once.call([this, use_page_cache, kept_in_memory, index_load_stats, io_ctx] {
+        return _load(use_page_cache, kept_in_memory, std::move(_meta_pb), index_load_stats, io_ctx);
     });
 }
 
 Status OrdinalIndexReader::_load(bool use_page_cache, bool kept_in_memory,
                                  std::unique_ptr<OrdinalIndexPB> index_meta,
-                                 OlapReaderStatistics* stats) {
+                                 OlapReaderStatistics* stats, const io::IOContext* io_ctx) {
     if (index_meta->root_page().is_root_data_page()) {
         // only one data page, no index page
         _num_pages = 1;
@@ -93,8 +94,10 @@ Status OrdinalIndexReader::_load(bool use_page_cache, bool kept_in_memory,
     // need to read index page
     OlapReaderStatistics tmp_stats;
     OlapReaderStatistics* stats_ptr = stats != nullptr ? stats : &tmp_stats;
-    PageReadOptions opts(io::IOContext {.is_index_data = true,
-                                        .file_cache_stats = &stats_ptr->file_cache_stats});
+    io::IOContext page_io_ctx = io_ctx != nullptr ? *io_ctx : io::IOContext {};
+    page_io_ctx.is_index_data = true;
+    page_io_ctx.file_cache_stats = &stats_ptr->file_cache_stats;
+    PageReadOptions opts(page_io_ctx);
     opts.use_page_cache = use_page_cache;
     opts.kept_in_memory = kept_in_memory;
     opts.type = INDEX_PAGE;
diff --git a/be/src/olap/rowset/segment_v2/ordinal_page_index.h b/be/src/olap/rowset/segment_v2/ordinal_page_index.h
index 6d85786ed17838..7898c0b873aa21 100644
--- a/be/src/olap/rowset/segment_v2/ordinal_page_index.h
+++ b/be/src/olap/rowset/segment_v2/ordinal_page_index.h
@@ -36,7 +36,8 @@ namespace doris {
 
 namespace io {
 class FileWriter;
-}
+struct IOContext;
+} // namespace io
 
 namespace segment_v2 {
 class ColumnIndexMetaPB;
@@ -75,7 +76,8 @@ class OrdinalIndexReader : public MetadataAdder<OrdinalIndexReader> {
     virtual ~OrdinalIndexReader();
 
     // load and parse the index page into memory
-    Status load(bool use_page_cache, bool kept_in_memory, OlapReaderStatistics* index_load_stats);
+    Status load(bool use_page_cache, bool kept_in_memory, OlapReaderStatistics* index_load_stats,
+                const io::IOContext* io_ctx = nullptr);
 
     // the returned iter points to the largest element which is less than `ordinal`,
     // or points to the first element if all elements are greater than `ordinal`,
@@ -94,8 +96,8 @@ class OrdinalIndexReader : public MetadataAdder<OrdinalIndexReader> {
 
 private:
     Status _load(bool use_page_cache, bool kept_in_memory,
-                 std::unique_ptr<OrdinalIndexPB> index_meta,
-                 OlapReaderStatistics* index_load_stats);
+                 std::unique_ptr<OrdinalIndexPB> index_meta, OlapReaderStatistics* index_load_stats,
+                 const io::IOContext* io_ctx);
 
 private:
     friend class OrdinalPageIndexIterator;
diff --git a/be/src/olap/rowset/segment_v2/segment.cpp b/be/src/olap/rowset/segment_v2/segment.cpp
index a0cf70d36b0f9f..5094a8ab4f4dfb 100644
--- a/be/src/olap/rowset/segment_v2/segment.cpp
+++ b/be/src/olap/rowset/segment_v2/segment.cpp
@@ -985,13 +985,14 @@ Status Segment::seek_and_read_by_rowid(const TabletSchema& schema, SlotDescripto
                                        uint32_t row_id, vectorized::MutableColumnPtr& result,
                                        StorageReadOptions& storage_read_options,
                                        std::unique_ptr<ColumnIterator>& iterator_hint) {
+    auto io_ctx = storage_read_options.io_ctx;
+    io_ctx.reader_type = ReaderType::READER_QUERY;
+    io_ctx.file_cache_stats = &storage_read_options.stats->file_cache_stats;
     segment_v2::ColumnIteratorOptions opt {
             .use_page_cache = !config::disable_storage_page_cache,
             .file_reader = file_reader().get(),
             .stats = storage_read_options.stats,
-            .io_ctx = io::IOContext {.reader_type = ReaderType::READER_QUERY,
-                                     .file_cache_stats =
-                                             &storage_read_options.stats->file_cache_stats},
+            .io_ctx = io_ctx,
     };
 
     std::vector<segment_v2::rowid_t> single_row_loc {row_id};
diff --git a/be/src/olap/rowset/segment_v2/zone_map_index.cpp b/be/src/olap/rowset/segment_v2/zone_map_index.cpp
index c80e7b9c91b770..3626efa235a935 100644
--- a/be/src/olap/rowset/segment_v2/zone_map_index.cpp
+++ b/be/src/olap/rowset/segment_v2/zone_map_index.cpp
@@ -182,20 +182,22 @@ Status TypedZoneMapIndexWriter<Type>::finish(io::FileWriter* file_writer,
 }
 
 Status ZoneMapIndexReader::load(bool use_page_cache, bool kept_in_memory,
-                                OlapReaderStatistics* index_load_stats) {
+                                OlapReaderStatistics* index_load_stats,
+                                const io::IOContext* io_ctx) {
     // TODO yyq: implement a new once flag to avoid status construct.
-    return _load_once.call([this, use_page_cache, kept_in_memory, index_load_stats] {
+    return _load_once.call([this, use_page_cache, kept_in_memory, index_load_stats, io_ctx] {
         return _load(use_page_cache, kept_in_memory, std::move(_page_zone_maps_meta),
-                     index_load_stats);
+                     index_load_stats, io_ctx);
     });
 }
 
 Status ZoneMapIndexReader::_load(bool use_page_cache, bool kept_in_memory,
                                  std::unique_ptr<IndexedColumnMetaPB> page_zone_maps_meta,
-                                 OlapReaderStatistics* index_load_stats) {
+                                 OlapReaderStatistics* index_load_stats,
+                                 const io::IOContext* io_ctx) {
     IndexedColumnReader reader(_file_reader, *page_zone_maps_meta);
-    RETURN_IF_ERROR(reader.load(use_page_cache, kept_in_memory, index_load_stats));
-    IndexedColumnIterator iter(&reader, index_load_stats);
+    RETURN_IF_ERROR(reader.load(use_page_cache, kept_in_memory, index_load_stats, io_ctx));
+    IndexedColumnIterator iter(&reader, index_load_stats, io_ctx);
 
     _page_zone_maps.resize(reader.num_values());
 
diff --git a/be/src/olap/rowset/segment_v2/zone_map_index.h b/be/src/olap/rowset/segment_v2/zone_map_index.h
index a5a0212777b8c8..ba05bc8f1d96f5 100644
--- a/be/src/olap/rowset/segment_v2/zone_map_index.h
+++ b/be/src/olap/rowset/segment_v2/zone_map_index.h
@@ -37,6 +37,7 @@ namespace doris {
 #include "common/compile_check_begin.h"
 namespace io {
 class FileWriter;
+struct IOContext;
 } // namespace io
 
 namespace segment_v2 {
@@ -167,7 +168,8 @@ class ZoneMapIndexReader : public MetadataAdder<ZoneMapIndexReader> {
 
     // load all page zone maps into memory
     Status load(bool use_page_cache, bool kept_in_memory,
-                OlapReaderStatistics* index_load_stats = nullptr);
+                OlapReaderStatistics* index_load_stats = nullptr,
+                const io::IOContext* io_ctx = nullptr);
 
     const std::vector<ZoneMapPB>& page_zone_maps() const { return _page_zone_maps; }
 
@@ -175,7 +177,7 @@ class ZoneMapIndexReader : public MetadataAdder<ZoneMapIndexReader> {
 
 private:
     Status _load(bool use_page_cache, bool kept_in_memory, std::unique_ptr<IndexedColumnMetaPB>,
-                 OlapReaderStatistics* index_load_stats);
+                 OlapReaderStatistics* index_load_stats, const io::IOContext* io_ctx);
 
     int64_t get_metadata_size() const override;
 
diff --git a/be/src/pipeline/exec/materialization_opertor.cpp b/be/src/pipeline/exec/materialization_opertor.cpp
index 1e3c34aabe89e3..dda061cbb45bde 100644
--- a/be/src/pipeline/exec/materialization_opertor.cpp
+++ b/be/src/pipeline/exec/materialization_opertor.cpp
@@ -21,13 +21,18 @@
 #include <fmt/format.h>
 #include <gen_cpp/internal_service.pb.h>
 
+#include <set>
+#include <sstream>
 #include <utility>
 
+#include "cloud/config.h"
+#include "common/config.h"
 #include "common/status.h"
 #include "exec/rowid_fetcher.h"
 #include "pipeline/exec/operator.h"
 #include "util/brpc_client_cache.h"
 #include "util/brpc_closure.h"
+#include "util/pretty_printer.h"
 #include "vec/columns/column.h"
 #include "vec/core/block.h"
 #include "vec/exec/scan/file_scanner.h"
@@ -35,6 +40,103 @@
 namespace doris {
 namespace pipeline {
 
+namespace {
+
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND =
+        "TopNLazyMaterializationSecondPhasePerBackend";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_ROWS_READ =
+        "TopNLazyMaterializationSecondPhasePerBackendRowsRead";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_SEGMENTS_READ =
+        "TopNLazyMaterializationSecondPhasePerBackendSegmentsRead";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_COUNT =
+        "TopNLazyMaterializationSecondPhasePerBackendLocalIOCount";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_BYTES =
+        "TopNLazyMaterializationSecondPhasePerBackendLocalIOBytes";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_COUNT =
+        "TopNLazyMaterializationSecondPhasePerBackendRemoteIOCount";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_BYTES =
+        "TopNLazyMaterializationSecondPhasePerBackendRemoteIOBytes";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_SKIP_CACHE_IO_COUNT =
+        "TopNLazyMaterializationSecondPhasePerBackendSkipCacheIOCount";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_BYTES =
+        "TopNLazyMaterializationSecondPhasePerBackendWriteCacheBytes";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_TIME =
+        "TopNLazyMaterializationSecondPhasePerBackendLocalIOTime";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_TIME =
+        "TopNLazyMaterializationSecondPhasePerBackendRemoteIOTime";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_IO_TIME =
+        "TopNLazyMaterializationSecondPhasePerBackendWriteCacheIOTime";
+
+void update_counter(RuntimeProfile* profile, const std::string& name, TUnit::type unit,
+                    int64_t value) {
+    COUNTER_UPDATE(ADD_COUNTER_WITH_LEVEL(profile, name, unit, 2), value);
+}
+
+void update_topn_lazy_materialization_profile(RuntimeProfile* profile,
+                                              const PTopNLazyMaterializationFileCacheStats& stats) {
+    if (profile == nullptr) {
+        return;
+    }
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOCount,
+                   TUnit::UNIT, stats.local_io_count());
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOBytes,
+                   TUnit::BYTES, stats.local_io_bytes());
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOCount,
+                   TUnit::UNIT, stats.remote_io_count());
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOBytes,
+                   TUnit::BYTES, stats.remote_io_bytes());
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseSkipCacheIOCount,
+                   TUnit::UNIT, stats.skip_cache_io_count());
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheBytes,
+                   TUnit::BYTES, stats.write_cache_bytes());
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOTime,
+                   TUnit::TIME_NS, stats.local_io_time());
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOTime,
+                   TUnit::TIME_NS, stats.remote_io_time());
+    update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheIOTime,
+                   TUnit::TIME_NS, stats.write_cache_io_time());
+}
+
+int64_t count_request_rows(const PMultiGetRequestV2& request) {
+    int64_t rows = 0;
+    for (const auto& request_block_desc : request.request_block_descs()) {
+        rows += request_block_desc.row_id_size();
+    }
+    return rows;
+}
+
+int64_t count_request_segments(const PMultiGetRequestV2& request) {
+    std::set<uint32_t> file_ids;
+    for (const auto& request_block_desc : request.request_block_descs()) {
+        DCHECK_EQ(request_block_desc.file_id_size(), request_block_desc.row_id_size());
+        for (const auto file_id : request_block_desc.file_id()) {
+            file_ids.insert(file_id);
+        }
+    }
+    return file_ids.size();
+}
+
+template <typename AppendValue>
+std::string format_array(size_t size, AppendValue append_value) {
+    std::stringstream values;
+    values << "[";
+    for (size_t i = 0; i < size; ++i) {
+        append_value(values, i);
+        values << ", ";
+    }
+    values << "]";
+    return values.str();
+}
+
+template <typename GetValue>
+std::string format_counter_array(size_t size, TUnit::type unit, GetValue get_value) {
+    return format_array(size, [&](std::stringstream& values, size_t i) {
+        values << PrettyPrinter::print(static_cast<int64_t>(get_value(i)), unit);
+    });
+}
+
+} // namespace
+
 void MaterializationSharedState::get_block(vectorized::Block* block) {
     for (int i = 0, j = 0, rowid_to_block_loc = rowid_locs[j]; i < origin_block.columns(); i++) {
         if (i != rowid_to_block_loc) {
@@ -54,6 +156,98 @@ void MaterializationSharedState::get_block(vectorized::Block* block) {
     origin_block.clear();
 }
 
+void MaterializationSharedState::_update_topn_lazy_materialization_profile(
+        RuntimeProfile* profile) {
+    DORIS_CHECK(profile != nullptr);
+    for (const auto& [backend_id, rpc_struct] : rpc_struct_map) {
+        const int64_t rows_read = count_request_rows(rpc_struct.request);
+        const int64_t segments_read = count_request_segments(rpc_struct.request);
+        update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseRowsRead,
+                       TUnit::UNIT, rows_read);
+        update_counter(profile, RowIdStorageReader::TopNLazyMaterializationSecondPhaseSegmentsRead,
+                       TUnit::UNIT, segments_read);
+
+        auto& stats = _topn_lazy_materialization_backend_stats[backend_id];
+        if (stats.backend.empty()) {
+            stats.backend = rpc_struct.backend_address.empty() ? fmt::format("id={}", backend_id)
+                                                               : rpc_struct.backend_address;
+        }
+        stats.rows_read += rows_read;
+        stats.segments_read += segments_read;
+        if (!rpc_struct.response.has_topn_lazy_materialization_file_cache_stats()) {
+            continue;
+        }
+
+        const auto& file_cache_stats =
+                rpc_struct.response.topn_lazy_materialization_file_cache_stats();
+        update_topn_lazy_materialization_profile(profile, file_cache_stats);
+        stats.local_io_count += file_cache_stats.local_io_count();
+        stats.local_io_bytes += file_cache_stats.local_io_bytes();
+        stats.remote_io_count += file_cache_stats.remote_io_count();
+        stats.remote_io_bytes += file_cache_stats.remote_io_bytes();
+        stats.skip_cache_io_count += file_cache_stats.skip_cache_io_count();
+        stats.write_cache_bytes += file_cache_stats.write_cache_bytes();
+        stats.local_io_time += file_cache_stats.local_io_time();
+        stats.remote_io_time += file_cache_stats.remote_io_time();
+        stats.write_cache_io_time += file_cache_stats.write_cache_io_time();
+    }
+
+    std::vector<const TopNLazyMaterializationBackendStats*> stats;
+    stats.reserve(_topn_lazy_materialization_backend_stats.size());
+    for (const auto& [_, backend_stats] : _topn_lazy_materialization_backend_stats) {
+        stats.push_back(&backend_stats);
+    }
+
+    const size_t size = stats.size();
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND,
+                             format_array(size, [&](std::stringstream& values, size_t i) {
+                                 values << stats[i]->backend;
+                             }));
+    profile->add_info_string(
+            TOPN_LAZY_MAT_PHASE2_PER_BACKEND_ROWS_READ,
+            format_counter_array(size, TUnit::UNIT, [&](size_t i) { return stats[i]->rows_read; }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_SEGMENTS_READ,
+                             format_counter_array(size, TUnit::UNIT, [&](size_t i) {
+                                 return stats[i]->segments_read;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_COUNT,
+                             format_counter_array(size, TUnit::UNIT, [&](size_t i) {
+                                 return stats[i]->local_io_count;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_BYTES,
+                             format_counter_array(size, TUnit::BYTES, [&](size_t i) {
+                                 return stats[i]->local_io_bytes;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_COUNT,
+                             format_counter_array(size, TUnit::UNIT, [&](size_t i) {
+                                 return stats[i]->remote_io_count;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_BYTES,
+                             format_counter_array(size, TUnit::BYTES, [&](size_t i) {
+                                 return stats[i]->remote_io_bytes;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_SKIP_CACHE_IO_COUNT,
+                             format_counter_array(size, TUnit::UNIT, [&](size_t i) {
+                                 return stats[i]->skip_cache_io_count;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_BYTES,
+                             format_counter_array(size, TUnit::BYTES, [&](size_t i) {
+                                 return stats[i]->write_cache_bytes;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_TIME,
+                             format_counter_array(size, TUnit::TIME_NS, [&](size_t i) {
+                                 return stats[i]->local_io_time;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_TIME,
+                             format_counter_array(size, TUnit::TIME_NS, [&](size_t i) {
+                                 return stats[i]->remote_io_time;
+                             }));
+    profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_IO_TIME,
+                             format_counter_array(size, TUnit::TIME_NS, [&](size_t i) {
+                                 return stats[i]->write_cache_io_time;
+                             }));
+}
+
 // Merges RPC responses from multiple BEs into `response_blocks` in the original row order.
 //
 // After parallel multiget_data_v2 RPCs complete, each BE's response contains a partial block
@@ -64,7 +258,9 @@ void MaterializationSharedState::get_block(vectorized::Block* block) {
 //   rpc_struct_map[backend_id].response  (per-BE partial blocks, unordered across BEs)
 //       + block_order_results[i][j]      (maps each output row → its source backend_id)
 //       → response_blocks[i]             (final merged result in original TopN row order)
-Status MaterializationSharedState::merge_multi_response() {
+Status MaterializationSharedState::merge_multi_response(RuntimeProfile* profile) {
+    _update_topn_lazy_materialization_profile(profile);
+
     // Outer loop: iterate over each relation (i.e., each rowid column / table).
     // A query with lazy materialization on 2 tables would have block_order_results.size() == 2,
     // each with its own set of response_blocks and RPC request_block_descs.
@@ -272,6 +468,9 @@ Status MaterializationSharedState::init_multi_requests(
     // Initialize the base struct of PMultiGetRequestV2
     multi_get_request.set_be_exec_version(state->be_exec_version());
     multi_get_request.set_wg_id(state->get_query_ctx()->workload_group()->id());
+    multi_get_request.set_file_cache_remote_only_on_miss(
+            config::is_cloud_mode() &&
+            state->query_options().enable_topn_lazy_mat_phase2_no_write_file_cache);
     auto* query_id = multi_get_request.mutable_query_id();
     query_id->set_hi(state->query_id().hi);
     query_id->set_lo(state->query_id().lo);
@@ -323,7 +522,10 @@ Status MaterializationSharedState::init_multi_requests(
                                FetchRpcStruct {.stub = std::move(client),
                                                .cntl = std::make_unique<brpc::Controller>(),
                                                .request = multi_get_request,
-                                               .response = PMultiGetResponseV2()});
+                                               .response = PMultiGetResponseV2(),
+                                               .backend_address = fmt::format(
+                                                       "id={} {}:{}", node_info.id, node_info.host,
+                                                       node_info.async_internal_port)});
     }
 
     return Status::OK();
@@ -433,7 +635,8 @@ Status MaterializationOperator::push(RuntimeState* state, vectorized::Block* in_
 
         if (local_state._materialization_state.need_merge_block) {
             SCOPED_TIMER(local_state._merge_response_timer);
-            RETURN_IF_ERROR(local_state._materialization_state.merge_multi_response());
+            RETURN_IF_ERROR(local_state._materialization_state.merge_multi_response(
+                    local_state.operator_profile()));
             local_state._max_rows_per_backend_counter->set(
                     (int64_t)local_state._materialization_state._max_rows_per_backend);
         }
diff --git a/be/src/pipeline/exec/materialization_opertor.h b/be/src/pipeline/exec/materialization_opertor.h
index a456374d09f939..8ad49f777a5687 100644
--- a/be/src/pipeline/exec/materialization_opertor.h
+++ b/be/src/pipeline/exec/materialization_opertor.h
@@ -19,6 +19,10 @@
 
 #include <stdint.h>
 
+#include <map>
+#include <string>
+#include <unordered_map>
+
 #include "common/status.h"
 #include "pipeline/exec/operator.h"
 
@@ -35,6 +39,7 @@ struct FetchRpcStruct {
     std::unique_ptr<brpc::Controller> cntl;
     PMultiGetRequestV2 request;
     PMultiGetResponseV2 response;
+    std::string backend_address;
 };
 
 struct MaterializationSharedState {
@@ -44,11 +49,27 @@ struct MaterializationSharedState {
     Status init_multi_requests(const TMaterializationNode& tnode, RuntimeState* state);
     Status create_muiltget_result(const vectorized::Columns& columns, bool eos, bool gc_id_map);
 
-    Status merge_multi_response();
+    Status merge_multi_response(RuntimeProfile* profile);
     void get_block(vectorized::Block* block);
 
 private:
     void _update_profile_info(int64_t backend_id, RuntimeProfile* response_profile);
+    void _update_topn_lazy_materialization_profile(RuntimeProfile* profile);
+
+    struct TopNLazyMaterializationBackendStats {
+        std::string backend;
+        int64_t rows_read = 0;
+        int64_t segments_read = 0;
+        int64_t local_io_count = 0;
+        int64_t local_io_bytes = 0;
+        int64_t remote_io_count = 0;
+        int64_t remote_io_bytes = 0;
+        int64_t skip_cache_io_count = 0;
+        int64_t write_cache_bytes = 0;
+        int64_t local_io_time = 0;
+        int64_t remote_io_time = 0;
+        int64_t write_cache_io_time = 0;
+    };
 
 public:
     bool rpc_struct_inited = false;
@@ -71,6 +92,10 @@ struct MaterializationSharedState {
     uint32_t _max_rows_per_backend = 0;
     // Store the number of rows processed by each backend
     std::unordered_map<int64_t, uint32_t> _backend_rows_count; // backend_id => rows_count
+
+private:
+    // backend id => accumulated TopN phase-2 profile stats.
+    std::map<int64_t, TopNLazyMaterializationBackendStats> _topn_lazy_materialization_backend_stats;
 };
 
 class MaterializationLocalState final : public PipelineXLocalState<FakeSharedState> {
diff --git a/be/test/io/cache/block_file_cache_test.cpp b/be/test/io/cache/block_file_cache_test.cpp
index 3f51263736ef99..00009986397f8d 100644
--- a/be/test/io/cache/block_file_cache_test.cpp
+++ b/be/test/io/cache/block_file_cache_test.cpp
@@ -209,6 +209,141 @@ void complete_into_memory(const io::FileBlocksHolder& holder) {
     }
 }
 
+TEST_F(BlockFileCacheTest, get_downloaded_blocks_if_fully_covered_is_read_only) {
+    const std::string local_cache_path =
+            (caches_dir / "remote_only_on_miss_helper_cache" / "").string();
+    if (fs::exists(local_cache_path)) {
+        fs::remove_all(local_cache_path);
+    }
+    fs::create_directories(local_cache_path);
+
+    io::BlockFileCache mgr(local_cache_path, cached_remote_reader_cache_settings());
+    ASSERT_TRUE(mgr.initialize().ok());
+    wait_until_cache_ready(mgr);
+
+    ReadStatistics read_stats;
+    io::CacheContext context;
+    context.stats = &read_stats;
+    context.cache_type = io::FileCacheType::NORMAL;
+    auto key = io::BlockFileCache::hash("remote_only_on_miss_helper_key");
+    const size_t block_size = cached_remote_reader_cache_settings().max_file_block_size;
+
+    auto assert_not_fully_covered = [&](size_t offset, size_t size, size_t expected_blocks_num) {
+        io::FileBlocks blocks;
+        bool fully_covered = true;
+        ASSERT_TRUE(mgr.get_downloaded_blocks_if_fully_covered(key, offset, size, context, &blocks,
+                                                               &fully_covered)
+                            .ok());
+        EXPECT_FALSE(fully_covered);
+        EXPECT_TRUE(blocks.empty());
+        EXPECT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), expected_blocks_num);
+    };
+
+    auto assert_fully_covered = [&](size_t offset, size_t size, size_t expected_blocks_size) {
+        io::FileBlocks blocks;
+        bool fully_covered = false;
+        ASSERT_TRUE(mgr.get_downloaded_blocks_if_fully_covered(key, offset, size, context, &blocks,
+                                                               &fully_covered)
+                            .ok());
+        EXPECT_TRUE(fully_covered);
+        EXPECT_EQ(blocks.size(), expected_blocks_size);
+    };
+
+    assert_not_fully_covered(0, 10, 0);
+
+    complete_into_memory(mgr.get_or_set(key, 0, block_size, context));
+    ASSERT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), 1);
+    assert_fully_covered(2, 5, 1);
+    assert_not_fully_covered(block_size - 4, 8, 1);
+
+    complete_into_memory(mgr.get_or_set(key, block_size, block_size, context));
+    ASSERT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), 2);
+    assert_fully_covered(block_size - 4, 8, 2);
+
+    complete_into_memory(mgr.get_or_set(key, 2 * block_size, block_size, context));
+    ASSERT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), 3);
+    assert_fully_covered(block_size - 4, block_size + 8, 3);
+
+    const size_t partial_block_size = 123 * 1024 + 17;
+    complete_into_memory(mgr.get_or_set(key, 3 * block_size, partial_block_size, context));
+    ASSERT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), 4);
+    assert_fully_covered(3 * block_size + 7, 31, 1);
+    assert_fully_covered(3 * block_size - 4, 16, 2);
+    assert_fully_covered(0, 3 * block_size + partial_block_size, 4);
+    assert_not_fully_covered(3 * block_size + partial_block_size - 4, 8, 4);
+    assert_not_fully_covered(0, 3 * block_size + partial_block_size + 1, 4);
+}
+
+TEST_F(BlockFileCacheTest, cached_remote_file_reader_remote_only_on_miss) {
+    const std::string local_cache_path =
+            (caches_dir / "remote_only_on_miss_reader_cache" / "").string();
+    BlockFileCache* cache = nullptr;
+    ASSERT_TRUE(create_cached_remote_reader_cache(local_cache_path, &cache).ok());
+
+    {
+        const auto remote_file = caches_dir / "remote_only_on_miss_reader_file";
+        {
+            std::ofstream ofs(remote_file, std::ios::binary | std::ios::trunc);
+            ASSERT_TRUE(ofs.is_open());
+            for (int i = 0; i < 2; ++i) {
+                std::string data(1_mb, static_cast<char>('a' + i));
+                ofs.write(data.data(), data.size());
+            }
+        }
+
+        FileReaderSPtr local_reader;
+        ASSERT_TRUE(global_local_filesystem()->open_file(remote_file, &local_reader).ok());
+        io::FileReaderOptions opts;
+        opts.cache_type = io::FileCachePolicy::FILE_BLOCK_CACHE;
+        opts.is_doris_table = true;
+        opts.tablet_id = 10086;
+        auto reader = std::make_shared<CachedRemoteFileReader>(local_reader, opts);
+
+        auto key = io::BlockFileCache::hash(remote_file.filename().string());
+        Defer fd_cache_cleanup {
+                [key] { io::FDCache::instance()->remove_file_reader(std::make_pair(key, 0)); }};
+        cache->remove_if_cached(key);
+
+        std::string buffer(4_kb, '\0');
+        size_t bytes_read = 0;
+        io::IOContext io_ctx;
+        io::FileCacheStatistics stats;
+        io_ctx.file_cache_stats = &stats;
+        io_ctx.file_cache_miss_policy = io::FileCacheMissPolicy::REMOTE_ONLY_ON_MISS;
+
+        ASSERT_TRUE(reader->read_at(123, Slice(buffer.data(), buffer.size()), &bytes_read, &io_ctx)
+                            .ok());
+        EXPECT_EQ(bytes_read, buffer.size());
+        EXPECT_EQ(stats.num_remote_io_total, 1);
+        EXPECT_EQ(stats.bytes_read_from_remote, buffer.size());
+        EXPECT_EQ(stats.num_skip_cache_io_total, 1);
+        EXPECT_EQ(stats.bytes_write_into_cache, 0);
+        EXPECT_TRUE(cache->get_blocks_by_key(key).empty());
+
+        io::FileCacheStatistics write_back_stats;
+        io_ctx.file_cache_stats = &write_back_stats;
+        io_ctx.file_cache_miss_policy = io::FileCacheMissPolicy::READ_THROUGH_AND_WRITE_BACK;
+        ASSERT_TRUE(reader->read_at(123, Slice(buffer.data(), buffer.size()), &bytes_read, &io_ctx)
+                            .ok());
+        EXPECT_EQ(bytes_read, buffer.size());
+        EXPECT_GT(write_back_stats.bytes_write_into_cache, 0);
+        EXPECT_FALSE(cache->get_blocks_by_key(key).empty());
+
+        io::FileCacheStatistics full_hit_stats;
+        io_ctx.file_cache_stats = &full_hit_stats;
+        io_ctx.file_cache_miss_policy = io::FileCacheMissPolicy::REMOTE_ONLY_ON_MISS;
+        ASSERT_TRUE(reader->read_at(123, Slice(buffer.data(), buffer.size()), &bytes_read, &io_ctx)
+                            .ok());
+        EXPECT_EQ(bytes_read, buffer.size());
+        EXPECT_EQ(full_hit_stats.num_local_io_total, 1);
+        EXPECT_EQ(full_hit_stats.bytes_read_from_local, buffer.size());
+        EXPECT_EQ(full_hit_stats.num_remote_io_total, 0);
+        EXPECT_EQ(full_hit_stats.bytes_write_into_cache, 0);
+    }
+
+    cleanup_cached_remote_reader_cache(local_cache_path);
+}
+
 void test_file_cache(io::FileCacheType cache_type) {
     TUniqueId query_id;
     query_id.hi = 1;
diff --git a/be/test/pipeline/operator/materialization_shared_state_test.cpp b/be/test/pipeline/operator/materialization_shared_state_test.cpp
index 769fc98e7869cd..b5814b42dc8cf1 100644
--- a/be/test/pipeline/operator/materialization_shared_state_test.cpp
+++ b/be/test/pipeline/operator/materialization_shared_state_test.cpp
@@ -19,6 +19,7 @@
 
 #include "pipeline/dependency.h"
 #include "pipeline/exec/materialization_opertor.h"
+#include "util/runtime_profile.h"
 #include "vec/columns/column_vector.h"
 #include "vec/core/field.h"
 #include "vec/data_types/data_type_number.h"
@@ -26,6 +27,15 @@
 
 namespace doris::pipeline {
 
+namespace {
+
+void add_request_row(PRequestBlockDesc* request_block_desc, uint32_t row_id, uint32_t file_id) {
+    request_block_desc->add_row_id(row_id);
+    request_block_desc->add_file_id(file_id);
+}
+
+} // namespace
+
 class MaterializationSharedStateTest : public testing::Test {
 protected:
     void SetUp() override {
@@ -107,13 +117,11 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponse) {
     // 2. Setup response blocks from multiple backends
     // Backend 1's response
     {
+        auto* request_block_desc =
+                _shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0);
+        add_request_row(request_block_desc, 0, 1);
+        add_request_row(request_block_desc, 1, 1);
         vectorized::Block resp_block1;
-        _shared_state->rpc_struct_map[_backend_id1]
-                .request.mutable_request_block_descs(0)
-                ->add_row_id(0);
-        _shared_state->rpc_struct_map[_backend_id1]
-                .request.mutable_request_block_descs(0)
-                ->add_row_id(1);
         auto resp_value_col1 = _int_type->create_column();
         auto* value_col_data1 = reinterpret_cast<vectorized::ColumnInt32*>(resp_value_col1.get());
         value_col_data1->insert(vectorized::Field::create_field<PrimitiveType::TYPE_INT>(100));
@@ -137,10 +145,10 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponse) {
 
     // Backend 2's response
     {
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(0),
+                2, 2);
         vectorized::Block resp_block2;
-        _shared_state->rpc_struct_map[_backend_id2]
-                .request.mutable_request_block_descs(0)
-                ->add_row_id(2);
         auto resp_value_col2 = _int_type->create_column();
         auto* value_col_data2 = reinterpret_cast<vectorized::ColumnInt32*>(resp_value_col2.get());
         value_col_data2->insert(vectorized::Field::create_field<PrimitiveType::TYPE_INT>(200));
@@ -166,7 +174,8 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponse) {
 
     // 4. Test merging responses
     vectorized::Block result_block;
-    Status st = _shared_state->merge_multi_response();
+    RuntimeProfile profile("MaterializationSharedStateTest");
+    Status st = _shared_state->merge_multi_response(&profile);
     _shared_state->get_block(&result_block);
     EXPECT_TRUE(st.ok());
 
@@ -219,10 +228,10 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseMultiBlocks) {
 
     // 2. Setup response blocks from multiple backends for first rowid
     {
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0),
+                0, 1);
         vectorized::Block resp_block1;
-        _shared_state->rpc_struct_map[_backend_id1]
-                .request.mutable_request_block_descs(0)
-                ->add_row_id(0);
         auto resp_value_col1 = _int_type->create_column();
         auto* value_col_data1 = reinterpret_cast<vectorized::ColumnInt32*>(resp_value_col1.get());
         value_col_data1->insert(vectorized::Field::create_field<PrimitiveType::TYPE_INT>(100));
@@ -244,10 +253,10 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseMultiBlocks) {
 
     // Backend 2's response for first rowid
     {
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(0),
+                0, 2);
         vectorized::Block resp_block2;
-        _shared_state->rpc_struct_map[_backend_id2]
-                .request.mutable_request_block_descs(0)
-                ->add_row_id(0);
         auto resp_value_col2 = _int_type->create_column();
         auto* value_col_data2 = reinterpret_cast<vectorized::ColumnInt32*>(resp_value_col2.get());
         value_col_data2->insert(vectorized::Field::create_field<PrimitiveType::TYPE_INT>(102));
@@ -270,10 +279,10 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseMultiBlocks) {
     _shared_state->rpc_struct_map[_backend_id1].request.add_request_block_descs();
     _shared_state->rpc_struct_map[_backend_id2].request.add_request_block_descs();
     {
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(1),
+                0, 3);
         vectorized::Block resp_block1;
-        _shared_state->rpc_struct_map[_backend_id1]
-                .request.mutable_request_block_descs(1)
-                ->add_row_id(0);
         auto resp_value_col1 = _int_type->create_column();
         auto* value_col_data1 = reinterpret_cast<vectorized::ColumnInt32*>(resp_value_col1.get());
         value_col_data1->insert(vectorized::Field::create_field<PrimitiveType::TYPE_INT>(200));
@@ -292,10 +301,10 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseMultiBlocks) {
     }
 
     {
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(1),
+                0, 4);
         vectorized::Block resp_block2;
-        _shared_state->rpc_struct_map[_backend_id2]
-                .request.mutable_request_block_descs(1)
-                ->add_row_id(0);
         auto resp_value_col2 = _int_type->create_column();
         auto* value_col_data2 = reinterpret_cast<vectorized::ColumnInt32*>(resp_value_col2.get());
         value_col_data2->insert(vectorized::Field::create_field<PrimitiveType::TYPE_INT>(201));
@@ -320,7 +329,8 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseMultiBlocks) {
 
     // 4. Test merging responses
     vectorized::Block result_block;
-    Status st = _shared_state->merge_multi_response();
+    RuntimeProfile profile("MaterializationSharedStateTest");
+    Status st = _shared_state->merge_multi_response(&profile);
     EXPECT_TRUE(st.ok());
     _shared_state->get_block(&result_block);
 
@@ -361,9 +371,9 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseBackendNotFound) {
 
     // --- BE_1: valid response with 1 row ---
     {
-        _shared_state->rpc_struct_map[_backend_id1]
-                .request.mutable_request_block_descs(0)
-                ->add_row_id(0);
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0),
+                0, 1);
         vectorized::Block resp_block;
         auto col = _int_type->create_column();
         reinterpret_cast<vectorized::ColumnInt32*>(col.get())->insert(
@@ -388,9 +398,9 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseBackendNotFound) {
     // After deserialization this produces a Block with 0 columns,
     // so is_empty_column() == true and it won't be inserted into block_maps.
     {
-        _shared_state->rpc_struct_map[_backend_id2]
-                .request.mutable_request_block_descs(0)
-                ->add_row_id(0);
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(0),
+                0, 2);
         PMultiGetResponseV2 response;
         response.add_blocks(); // empty PMultiGetBlockV2, no mutable_block() data
         _shared_state->rpc_struct_map[_backend_id2].response = std::move(response);
@@ -410,7 +420,8 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseBackendNotFound) {
     _shared_state->rowid_locs = {0};
 
     // merge_multi_response() should return InternalError
-    Status st = _shared_state->merge_multi_response();
+    RuntimeProfile profile("MaterializationSharedStateTest");
+    Status st = _shared_state->merge_multi_response(&profile);
     ASSERT_FALSE(st.ok());
     ASSERT_TRUE(st.is<ErrorCode::INTERNAL_ERROR>());
     ASSERT_TRUE(st.to_string().find("not match request row id count") != std::string::npos)
@@ -444,9 +455,9 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseStaleBlockMaps) {
 
     // --- Build BE_1's response: blocks[0]=1 row (INT), blocks[1]=empty ---
     {
-        _shared_state->rpc_struct_map[_backend_id1]
-                .request.mutable_request_block_descs(0)
-                ->add_row_id(0);
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0),
+                0, 1);
         PMultiGetResponseV2 response;
 
         // blocks[0]: 1 row of INT for relation 0
@@ -481,9 +492,9 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseStaleBlockMaps) {
         reinterpret_cast<vectorized::ColumnString*>(col.get())->insert_data("Alice", 5);
         rel1_block.insert({make_nullable(std::move(col)), make_nullable(_string_type), "name"});
 
-        _shared_state->rpc_struct_map[_backend_id2]
-                .request.mutable_request_block_descs(1)
-                ->add_row_id(0);
+        add_request_row(
+                _shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(1),
+                0, 2);
         auto* pb1 = response.add_blocks()->mutable_block();
         size_t us = 0, cs = 0;
         int64_t ct = 0;
@@ -510,7 +521,8 @@ TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseStaleBlockMaps) {
     _shared_state->rowid_locs = {0, 1};
 
     // merge should succeed — each relation only references the BE that has data
-    Status st = _shared_state->merge_multi_response();
+    RuntimeProfile profile("MaterializationSharedStateTest");
+    Status st = _shared_state->merge_multi_response(&profile);
     ASSERT_TRUE(st.ok()) << "merge_multi_response failed: " << st.to_string();
 
     // Verify results
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 f055133b6a3f30..80c89a3fba5f95 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
@@ -478,6 +478,9 @@ public class SessionVariable implements Serializable, Writable {
 
     public static final String DISABLE_FILE_CACHE = "disable_file_cache";
 
+    public static final String ENABLE_TOPN_LAZY_MAT_PHASE2_NO_WRITE_FILE_CACHE
+            = "enable_topn_lazy_mat_phase2_no_write_file_cache";
+
     public static final String FILE_CACHE_QUERY_LIMIT_PERCENT = "file_cache_query_limit_percent";
 
     public static final String FILE_CACHE_BASE_PATH = "file_cache_base_path";
@@ -2111,6 +2114,14 @@ public boolean isEnableHboNonStrictMatchingMode() {
     @VariableMgr.VarAttr(name = DISABLE_FILE_CACHE, needForward = true)
     public boolean disableFileCache = false;
 
+    @VariableMgr.VarAttr(name = ENABLE_TOPN_LAZY_MAT_PHASE2_NO_WRITE_FILE_CACHE, needForward = true,
+            description = {
+                    "开启后，TopN 延迟物化第二阶段读取在 file cache miss 时直接读远端且不写回 file cache。",
+                    "When enabled, TopN lazy materialization phase-2 reads go remote-only on "
+                            + "file-cache miss and do not write the missed range back to file cache."
+            })
+    public boolean enableTopnLazyMatPhase2NoWriteFileCache = false;
+
     // Whether enable block file cache. Only take effect when BE config item enable_file_cache is true.
     @VariableMgr.VarAttr(name = ENABLE_FILE_CACHE, needForward = true, description = {
             "是否启用 file cache。该变量只有在 be.conf 中 enable_file_cache=true 时才有效，"
@@ -5054,6 +5065,7 @@ public TQueryOptions toThrift() {
         tResult.setOptimizeIndexScanParallelism(optimizeIndexScanParallelism);
         tResult.setSkipBadTablet(skipBadTablet);
         tResult.setDisableFileCache(disableFileCache);
+        tResult.setEnableTopnLazyMatPhase2NoWriteFileCache(enableTopnLazyMatPhase2NoWriteFileCache);
 
         tResult.setEnablePreferCachedRowset(getEnablePreferCachedRowset());
         tResult.setQueryFreshnessToleranceMs(getQueryFreshnessToleranceMs());
diff --git a/gensrc/proto/internal_service.proto b/gensrc/proto/internal_service.proto
index f3af424a3467a6..716b5a78f8079a 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -781,6 +781,18 @@ message PRequestBlockDesc {
     repeated uint32 column_idxs = 7;
 }
 
+message PTopNLazyMaterializationFileCacheStats {
+    optional int64 local_io_count = 1;
+    optional int64 local_io_bytes = 2;
+    optional int64 remote_io_count = 3;
+    optional int64 remote_io_bytes = 4;
+    optional int64 skip_cache_io_count = 5;
+    optional int64 write_cache_bytes = 6;
+    optional int64 local_io_time = 7;
+    optional int64 remote_io_time = 8;
+    optional int64 write_cache_io_time = 9;
+}
+
 message PMultiGetRequestV2 {
     repeated PRequestBlockDesc request_block_descs = 1;
 
@@ -789,6 +801,7 @@ message PMultiGetRequestV2 {
     optional PUniqueId query_id = 3;
     optional bool gc_id_map = 4;
     optional uint64 wg_id = 5;
+    optional bool file_cache_remote_only_on_miss = 6;
 };
 
 message PMultiGetBlockV2 {
@@ -805,6 +818,7 @@ message PMultiGetBlockV2 {
 message PMultiGetResponseV2 {
     optional PStatus status = 1;
     repeated PMultiGetBlockV2 blocks = 2;
+    optional PTopNLazyMaterializationFileCacheStats topn_lazy_materialization_file_cache_stats = 3;
 };
 
 message PFetchColIdsRequest {
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index eeaf3f12a23eb2..215c0fc54f2a3c 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -432,6 +432,7 @@ struct TQueryOptions {
   // In read path, read from file cache or remote storage when execute query.
   1000: optional bool disable_file_cache = false
   1001: optional i32 file_cache_query_limit_percent = -1
+  1003: optional bool enable_topn_lazy_mat_phase2_no_write_file_cache = false
 }
 
 
diff --git a/regression-test/suites/cloud_p0/cache/topn_lazy_file_cache/test_topn_lazy_mat_phase2_no_write_file_cache.groovy b/regression-test/suites/cloud_p0/cache/topn_lazy_file_cache/test_topn_lazy_mat_phase2_no_write_file_cache.groovy
new file mode 100644
index 00000000000000..2e3ea83265c973
--- /dev/null
+++ b/regression-test/suites/cloud_p0/cache/topn_lazy_file_cache/test_topn_lazy_mat_phase2_no_write_file_cache.groovy
@@ -0,0 +1,294 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.Http
+
+import java.util.regex.Pattern
+
+suite("test_topn_lazy_mat_phase2_no_write_file_cache", "docker") {
+    def options = new ClusterOptions()
+    options.feNum = 1
+    options.beNum = 1
+    options.msNum = 1
+    options.cloudMode = true
+    options.beConfigs += [
+        "enable_file_cache=true",
+        "disable_storage_page_cache=true",
+        "enable_java_support=false",
+        "enable_evict_file_cache_in_advance=false",
+        "file_cache_enter_disk_resource_limit_mode_percent=99",
+        "file_cache_each_block_size=4096",
+        "file_cache_path=[{\"path\":\"/opt/apache-doris/be/storage/file_cache\","
+                + "\"total_size\":104857600,\"query_limit\":104857600}]"
+    ]
+
+    docker(options) {
+        def clusters = sql "SHOW CLUSTERS"
+        assert !clusters.isEmpty()
+        def computeGroup = clusters[0][0]
+        sql "use @${computeGroup}"
+
+        def backends = sql "SHOW BACKENDS"
+        assert backends.size() == 1
+        def beHost = backends[0][1]
+        def beHttpPort = backends[0][4]
+        def clearFileCache = {
+            def result = Http.GET("http://${beHost}:${beHttpPort}/api/file_cache?op=clear&sync=true", true)
+            logger.info("clear file cache result: ${result}")
+        }
+
+        sql "set enable_profile = true"
+        sql "set profile_level = 2"
+        sql "set enable_sql_cache = false"
+        sql "set enable_query_cache = false"
+        sql "set enable_page_cache = false"
+        sql "set topn_lazy_materialization_threshold = 1024"
+
+        def metricValue = { String profileText, String metricName ->
+            def matcher = profileText =~
+                    /(?m)^\s*-\s*${Pattern.quote(metricName)}:\s*(?:sum\s+)?([0-9]+(?:\.[0-9]+)?).*$/
+            assert matcher.find() : "missing metric ${metricName} in profile:\n${profileText}"
+            return new BigDecimal(matcher.group(1))
+        }
+
+        def metricValues = { String profileText, String metricName ->
+            def matcher = profileText =~
+                    /(?m)^\s*-\s*${Pattern.quote(metricName)}:\s*(?:sum\s+)?([0-9]+(?:\.[0-9]+)?).*$/
+            def values = []
+            while (matcher.find()) {
+                values.add(new BigDecimal(matcher.group(1)))
+            }
+            assert !values.isEmpty() : "missing metric ${metricName} in profile:\n${profileText}"
+            return values
+        }
+
+        def numericValues = { String text ->
+            def matcher = text =~ /([0-9]+(?:\.[0-9]+)?)/
+            def values = []
+            while (matcher.find()) {
+                values.add(new BigDecimal(matcher.group(1)))
+            }
+            return values
+        }
+
+        def metricLineValues = { String profileText, String metricName ->
+            def matcher = profileText =~ /(?m)^\s*-\s*${Pattern.quote(metricName)}:\s*(.*)$/
+            def values = []
+            while (matcher.find()) {
+                values.add(matcher.group(1).trim())
+            }
+            return values
+        }
+
+        def topnSecondPhaseMetricNames = [
+            "TopNLazyMaterializationSecondPhaseLocalIOCount",
+            "TopNLazyMaterializationSecondPhaseLocalIOBytes",
+            "TopNLazyMaterializationSecondPhaseRemoteIOCount",
+            "TopNLazyMaterializationSecondPhaseRemoteIOBytes",
+            "TopNLazyMaterializationSecondPhaseSkipCacheIOCount",
+            "TopNLazyMaterializationSecondPhaseWriteCacheBytes",
+            "TopNLazyMaterializationSecondPhaseLocalIOTime",
+            "TopNLazyMaterializationSecondPhaseRemoteIOTime",
+            "TopNLazyMaterializationSecondPhaseWriteCacheIOTime",
+            "TopNLazyMaterializationSecondPhaseRowsRead",
+            "TopNLazyMaterializationSecondPhaseSegmentsRead",
+            "TopNLazyMaterializationSecondPhasePerBackend",
+            "TopNLazyMaterializationSecondPhasePerBackendRowsRead",
+            "TopNLazyMaterializationSecondPhasePerBackendSegmentsRead",
+            "TopNLazyMaterializationSecondPhasePerBackendLocalIOCount",
+            "TopNLazyMaterializationSecondPhasePerBackendLocalIOBytes",
+            "TopNLazyMaterializationSecondPhasePerBackendRemoteIOCount",
+            "TopNLazyMaterializationSecondPhasePerBackendRemoteIOBytes",
+            "TopNLazyMaterializationSecondPhasePerBackendSkipCacheIOCount",
+            "TopNLazyMaterializationSecondPhasePerBackendWriteCacheBytes",
+            "TopNLazyMaterializationSecondPhasePerBackendLocalIOTime",
+            "TopNLazyMaterializationSecondPhasePerBackendRemoteIOTime",
+            "TopNLazyMaterializationSecondPhasePerBackendWriteCacheIOTime"
+        ]
+
+        def logTopnSecondPhaseMetrics = { String name, String profileText ->
+            def metrics = topnSecondPhaseMetricNames.collectEntries { metricName ->
+                def values = metricLineValues(profileText, metricName)
+                assert !values.isEmpty() : "missing metric ${metricName} in profile:\n${profileText}"
+                [(metricName): values]
+            }
+            logger.info("${name} TopN lazy materialization second phase metrics: ${metrics}")
+            logger.info("${name} CachedPagesNum values: ${metricLineValues(profileText, 'CachedPagesNum')}")
+        }
+
+        def runProfileQuery = { String name, String query, Closure checker ->
+            profile(name) {
+                run {
+                    sql "/* ${name} */ ${query}"
+                    sleep(1000)
+                }
+                check { profileString, exception ->
+                    if (exception != null) {
+                        logger.error("Profile failed, profile result:\n${profileString}", exception)
+                        throw exception
+                    }
+                    assert profileString.contains("TopNLazyMaterializationSecondPhase") :
+                            "missing TopN lazy materialization profile:\n${profileString}"
+                    assert profileString.contains("Is  Cached:  No")
+                            || profileString.contains("Is Cached: No") :
+                            "query should not hit SQL/query cache:\n${profileString}"
+                    logTopnSecondPhaseMetrics(name, profileString)
+                    checker(profileString)
+                }
+            }
+        }
+
+        def assertRemoteOnlyMiss = { String profileText ->
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseRemoteIOCount") > 0
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseRemoteIOBytes") > 0
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseSkipCacheIOCount") > 0
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseWriteCacheBytes")
+                    .compareTo(BigDecimal.ZERO) == 0
+        }
+
+        def assertPerBackendRowsMatchAggregate = { String profileText ->
+            def aggregateRows = metricValues(profileText,
+                    "TopNLazyMaterializationSecondPhaseRowsRead")
+                    .inject(BigDecimal.ZERO) { sum, value -> sum + value }
+            def perBackendRows = metricLineValues(profileText,
+                    "TopNLazyMaterializationSecondPhasePerBackendRowsRead")
+                    .collectMany { line -> numericValues(line) }
+            assert !perBackendRows.isEmpty() :
+                    "missing per-backend rows-read values:\n${profileText}"
+            def perBackendRowsSum = perBackendRows.inject(BigDecimal.ZERO) { sum, value -> sum + value }
+            assert aggregateRows.compareTo(perBackendRowsSum) == 0 :
+                    "per-backend rows-read sum ${perBackendRowsSum} should equal aggregate ${aggregateRows}:\n${profileText}"
+        }
+
+        def assertLocalFullHit = { String profileText ->
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseLocalIOCount") > 0
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseLocalIOBytes") > 0
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseRemoteIOCount")
+                    .compareTo(BigDecimal.ZERO) == 0
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseRemoteIOBytes")
+                    .compareTo(BigDecimal.ZERO) == 0
+            assert metricValue(profileText, "TopNLazyMaterializationSecondPhaseWriteCacheBytes")
+                    .compareTo(BigDecimal.ZERO) == 0
+        }
+
+        sql "DROP TABLE IF EXISTS topn_lazy_remote_only_no_row_store"
+        sql """
+            CREATE TABLE topn_lazy_remote_only_no_row_store (
+                k INT NOT NULL,
+                sort_key INT NOT NULL,
+                payload VARCHAR(4096) NOT NULL,
+                pad VARCHAR(4096) NOT NULL
+            ) ENGINE=OLAP
+            DUPLICATE KEY(k)
+            DISTRIBUTED BY HASH(k) BUCKETS 1
+            PROPERTIES (
+                "replication_num" = "1",
+                "disable_auto_compaction" = "true",
+                "store_row_column" = "false"
+            )
+        """
+        sql """
+            INSERT INTO topn_lazy_remote_only_no_row_store
+            SELECT number,
+                   4096 - number,
+                   repeat(cast(number as string), 128),
+                   repeat('x', 256)
+            FROM numbers("number" = "4096")
+        """
+
+        def noRowStoreQuery =
+                "SELECT k, payload, pad FROM topn_lazy_remote_only_no_row_store ORDER BY sort_key LIMIT 16"
+        explain {
+            sql noRowStoreQuery
+            contains("MaterializeNode")
+        }
+
+        clearFileCache()
+        sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+        runProfileQuery("topn_lazy_remote_only_no_row_store_remote_only_miss",
+                noRowStoreQuery, assertRemoteOnlyMiss)
+
+        clearFileCache()
+        sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = false"
+        sql "/* topn_lazy_remote_only_no_row_store_local_full_hit_warm */ ${noRowStoreQuery}"
+        sleep(1000)
+
+        sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+        runProfileQuery("topn_lazy_remote_only_no_row_store_local_full_hit",
+                noRowStoreQuery, assertLocalFullHit)
+
+        clearFileCache()
+        sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+        sql "set batch_size = 1"
+        try {
+            runProfileQuery("topn_lazy_remote_only_no_row_store_multi_fetch_accumulate",
+                    noRowStoreQuery, { profileText ->
+                        assertRemoteOnlyMiss(profileText)
+                        assertPerBackendRowsMatchAggregate(profileText)
+                    })
+        } finally {
+            sql "set batch_size = 4062"
+        }
+
+        sql "DROP TABLE IF EXISTS topn_lazy_remote_only_row_store"
+        sql """
+            CREATE TABLE topn_lazy_remote_only_row_store (
+                k INT NOT NULL,
+                sort_key INT NOT NULL,
+                payload VARCHAR(4096) NOT NULL,
+                pad VARCHAR(4096) NOT NULL
+            ) ENGINE=OLAP
+            DUPLICATE KEY(k)
+            DISTRIBUTED BY HASH(k) BUCKETS 1
+            PROPERTIES (
+                "replication_num" = "1",
+                "disable_auto_compaction" = "true",
+                "store_row_column" = "true"
+            )
+        """
+        sql """
+            INSERT INTO topn_lazy_remote_only_row_store
+            SELECT number,
+                   4096 - number,
+                   repeat(cast(number as string), 128),
+                   repeat('x', 256)
+            FROM numbers("number" = "4096")
+        """
+
+        def rowStoreQuery =
+                "SELECT k, payload, pad FROM topn_lazy_remote_only_row_store ORDER BY sort_key LIMIT 16"
+        explain {
+            sql rowStoreQuery
+            contains("MaterializeNode")
+        }
+
+        clearFileCache()
+        sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+        runProfileQuery("topn_lazy_remote_only_row_store_remote_only_miss",
+                rowStoreQuery, assertRemoteOnlyMiss)
+
+        clearFileCache()
+        sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = false"
+        sql "/* topn_lazy_remote_only_row_store_local_full_hit_warm */ ${rowStoreQuery}"
+        sleep(1000)
+
+        sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+        runProfileQuery("topn_lazy_remote_only_row_store_local_full_hit",
+                rowStoreQuery, assertLocalFullHit)
+    }
+}