branch-4.0: [improvement](topn) Add option to skip file cache writes in topn lazy materialization

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(

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,

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);
 }

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
      */