[refactor](be) Align file cache probe results with read blocks

### What problem does this PR solve?

Issue Number: None

Related PR: apache#65658

Problem Summary: The async cache-write planner queried one block-aligned range, but `BlockFileCache::probe` returned a `FileBlocksHolder` of cache hits plus an independent gap list. The planner then had to scan all hits and gaps for every logical read block even though the read-plan blocks and probe slots use the same block boundaries. This obscured the alignment invariant and introduced unnecessary nested matching logic on the query read path.

Change the probe contract to return one ordered nullable `FileBlock` pointer per aligned input block. A non-null slot is asserted to have the exact corresponding range, except that the final block may end at EOF, while a null slot directly represents a cache miss. The planner now preserves its inflight-first fast path and joins probe results to plan blocks by index; materialization also reads the matching slot directly.

Remove `FileBlocksHolder` and the independent gaps from `FileBlocksProbeResult`. Preserve the existing deferred cleanup semantics for EMPTY and deleting cache blocks through a shared cache-user reference release helper rather than embedding a holder in the probe result. Update focused and end-to-end tests for hit/miss slots, a short final block, retained block states, self-heal cleanup, and an aligned direct-cache prefix followed by an async-written suffix.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - 51 related ASAN BE unit tests passed with `run-be-ut.sh` and `-j100`
    - 2 focused probe/direct-prefix ASAN BE unit tests passed with `run-be-ut.sh` and `-j100`
    - `build-support/check-format.sh` passed
- Behavior changed: No; this simplifies an internal probe/planning contract without changing user-visible cache semantics
- Does this need documentation: No

Author: bobhan1

be/src/io/cache/block_file_cache.cpp

@@ -845,9 +845,9 @@ Status BlockFileCache::get_downloaded_blocks_if_fully_covered(const UInt128Wrapp
 FileBlocksProbeResult BlockFileCache::probe(const UInt128Wrapper& hash, size_t offset, size_t size,
                                             const CacheContext& context) {
     DORIS_CHECK(size > 0);
-    const FileBlock::Range range(offset, offset + size - 1);
-    FileBlocks blocks;
-    std::vector<FileBlock::Range> gaps;
+    DORIS_CHECK(_max_file_block_size > 0);
+    const size_t end = offset + size;
+    DORIS_CHECK(end > offset);
     std::lock_guard cache_lock(_mutex);
 
     auto file_iterator = _files.find(hash);
@@ -860,46 +860,36 @@ FileBlocksProbeResult BlockFileCache::probe(const UInt128Wrapper& hash, size_t o
         _storage->load_blocks_directly_unlocked(this, key, cache_lock);
         file_iterator = _files.find(hash);
     }
-    if (file_iterator == _files.end()) {
-        gaps.emplace_back(range);
-        return FileBlocksProbeResult(std::move(blocks), std::move(gaps));
-    }
 
-    auto& file_blocks = file_iterator->second;
-    DORIS_CHECK(!file_blocks.empty());
-    auto block_iterator = file_blocks.lower_bound(range.left);
-    if (block_iterator != file_blocks.begin()) {
-        auto previous = std::prev(block_iterator);
-        if (previous->second.file_block->range().right >= range.left) {
-            block_iterator = previous;
+    FileBlocksByOffset* cached_blocks = nullptr;
+    FileBlocksByOffset::iterator cached_block;
+    if (file_iterator != _files.end()) {
+        DORIS_CHECK(!file_iterator->second.empty());
+        cached_blocks = &file_iterator->second;
+        cached_block = cached_blocks->lower_bound(offset);
+        if (cached_block != cached_blocks->begin()) {
+            const auto& previous_range = std::prev(cached_block)->second.file_block->range();
+            DORIS_CHECK(previous_range.right < offset);
         }
     }
 
-    size_t current = range.left;
-    for (; block_iterator != file_blocks.end(); ++block_iterator) {
-        const auto& block = block_iterator->second.file_block;
-        const auto& block_range = block->range();
-        if (block_range.left > range.right) {
-            break;
-        }
-        if (block_range.right < range.left) {
-            continue;
+    std::vector<FileBlockSPtr> result;
+    result.reserve(size / _max_file_block_size + (size % _max_file_block_size != 0));
+    for (size_t block_offset = offset; block_offset < end;) {
+        const size_t block_size = std::min(_max_file_block_size, end - block_offset);
+        const FileBlock::Range expected_range(block_offset, block_offset + block_size - 1);
+        FileBlockSPtr file_block;
+        if (cached_blocks != nullptr && cached_block != cached_blocks->end() &&
+            cached_block->second.file_block->range().left <= expected_range.right) {
+            file_block = cached_block->second.file_block;
+            DORIS_CHECK(file_block->range().left == expected_range.left);
+            DORIS_CHECK(file_block->range().right == expected_range.right);
+            ++cached_block;
         }
-        const size_t clipped_left = std::max(block_range.left, range.left);
-        const size_t clipped_right = std::min(block_range.right, range.right);
-        if (current < clipped_left) {
-            gaps.emplace_back(current, clipped_left - 1);
-        }
-        blocks.emplace_back(block);
-        current = clipped_right + 1;
-        if (current > range.right) {
-            break;
-        }
-    }
-    if (current <= range.right) {
-        gaps.emplace_back(current, range.right);
+        result.emplace_back(std::move(file_block));
+        block_offset += block_size;
     }
-    return FileBlocksProbeResult(std::move(blocks), std::move(gaps));
+    return FileBlocksProbeResult(std::move(result));
 }
 
 void BlockFileCache::touch_probe_block_if_cached(const FileBlockSPtr& block,

be/src/io/cache/block_file_cache.h

@@ -26,6 +26,7 @@
 #include <boost/lockfree/spsc_queue.hpp>
 #include <functional>
 #include <limits>
+#include <map>
 #include <memory>
 #include <mutex>
 #include <optional>
@@ -83,15 +84,17 @@ class LockScopedTimer {
 class FSFileCacheStorage;
 
 struct FileBlocksProbeResult {
-    FileBlocksProbeResult(FileBlocks blocks, std::vector<FileBlock::Range> gaps_)
-            : holder(std::move(blocks)), gaps(std::move(gaps_)) {}
+    explicit FileBlocksProbeResult(std::vector<FileBlockSPtr> file_blocks_)
+            : file_blocks(std::move(file_blocks_)) {}
     FileBlocksProbeResult(FileBlocksProbeResult&&) noexcept = default;
     FileBlocksProbeResult& operator=(FileBlocksProbeResult&&) noexcept = delete;
     FileBlocksProbeResult(const FileBlocksProbeResult&) = delete;
     FileBlocksProbeResult& operator=(const FileBlocksProbeResult&) = delete;
+    ~FileBlocksProbeResult();
 
-    FileBlocksHolder holder;
-    std::vector<FileBlock::Range> gaps;
+    /// One entry per cache-block-sized input slot, in offset order. A null entry is a cache miss;
+    /// a non-null entry has the exact slot range, except that the final slot may end at EOF.
+    std::vector<FileBlockSPtr> file_blocks;
 };
 
 // NeedUpdateLRUBlocks keeps FileBlockSPtr entries that require LRU updates in a
@@ -254,8 +257,10 @@ class BlockFileCache {
     FileBlocksHolder get_or_set(const UInt128Wrapper& hash, size_t offset, size_t size,
                                 CacheContext& context);
 
-    /// Return existing blocks and uncovered gaps for `[offset, offset + size)` without creating
-    /// cache cells or touching LRU state. `context` supplies cache-type visibility rules.
+    /// Probe the block-aligned `[offset, offset + size)` range without creating cache cells or
+    /// touching LRU state. The result contains one ordered slot per cache block; each slot is null
+    /// on miss or owns the exactly aligned existing block. `context` supplies cache metadata when
+    /// lazy loading is required.
     FileBlocksProbeResult probe(const UInt128Wrapper& hash, size_t offset, size_t size,
                                 const CacheContext& context);
 

be/src/io/cache/cached_remote_file_reader.h

@@ -170,33 +170,34 @@ class CachedRemoteFileReader final : public FileReader,
 
     /// Build block-aligned source coverage for the unread suffix. It first performs one inflight
     /// batch lookup; a fully covered request returns without taking the BlockFileCache lock, while
-    /// incomplete coverage triggers one read-only cache probe for the whole aligned range.
+    /// incomplete coverage triggers one read-only whole-range probe with one result per aligned
+    /// block.
     /// @param[in] remaining_offset First user byte not filled by the direct-cache path.
     /// @param[in] remaining_size Number of unread user bytes.
     /// @param[in] write_epoch Epoch captured before any lookup or remote IO.
     /// @param[in] io_ctx Context used to build the cache admission/probe context.
     /// @param[in,out] stats Lookup and probe counters updated during planning.
-    /// @return Plan that owns any required probe holder and first-to-last remote range.
+    /// @return Plan that owns any retained probe blocks and first-to-last remote range.
     AsyncReadPlan _build_async_read_plan(size_t remaining_offset, size_t remaining_size,
                                          uint64_t write_epoch, const IOContext* io_ctx,
                                          ReadStatistics& stats);
 
     /// Copy one block already available from an inflight buffer or downloaded cache file. Cache
     /// state is revalidated before IO; a race is reported to the caller as a simple full-range
     /// remote fallback.
-    /// @param[in] plan Plan owning the probe holder and user boundaries.
-    /// @param[in] read_block Aligned block classified as INFLIGHT, CACHE, or DOWNLOADING.
+    /// @param[in] plan Plan owning the probed blocks and user boundaries.
+    /// @param[in] block_index Index of the aligned block and its matching probe result.
     /// @param[in] user_offset Original user request offset used to locate the destination slice.
     /// @param[out] result Destination buffer for the complete user request.
     /// @param[in] cache_context Context used only when a successful local read touches LRU.
     /// @param[in,out] stats Local-read timing counters.
     /// @param[in,out] materialized_bytes User bytes copied from cache or inflight memory.
     /// @param[in,out] need_self_heal Set when a cache file disappears during a local read.
     /// @return true when the block was copied; false when the caller should use remote fallback.
-    bool _materialize_async_block(const AsyncReadPlan& plan, const AsyncReadBlock& read_block,
-                                  size_t user_offset, Slice result,
-                                  const CacheContext& cache_context, ReadStatistics& stats,
-                                  size_t* materialized_bytes, bool* need_self_heal);
+    bool _materialize_async_block(const AsyncReadPlan& plan, size_t block_index, size_t user_offset,
+                                  Slice result, const CacheContext& cache_context,
+                                  ReadStatistics& stats, size_t* materialized_bytes,
+                                  bool* need_self_heal);
 
     /// Copy only blocks before the first and after the last REMOTE block. When no REMOTE block
     /// exists, copy the entire request. DOWNLOADING blocks outside the remote span keep the

be/src/io/cache/cached_remote_file_reader_async_write.cpp

@@ -122,8 +122,7 @@ CacheWriteMode CachedRemoteFileReader::_resolve_cache_write_mode(const IOContext
 
 // Build a deliberately small plan. The inflight index is checked first so a fully covered read
 // avoids BlockFileCache::probe and its cache mutex. Only an incomplete inflight lookup performs one
-// whole-range cache probe; the per-block scans favor readability because read_at normally spans
-// very few cache blocks.
+// whole-range probe whose result entries map directly to the logical plan blocks.
 CachedRemoteFileReader::AsyncReadPlan CachedRemoteFileReader::_build_async_read_plan(
         size_t remaining_offset, size_t remaining_size, uint64_t write_epoch,
         const IOContext* io_ctx, ReadStatistics& stats) {
@@ -140,7 +139,8 @@ CachedRemoteFileReader::AsyncReadPlan CachedRemoteFileReader::_build_async_read_
          block_offset += cache_block_size) {
         const size_t block_size = std::min(cache_block_size, size() - block_offset);
         DORIS_CHECK(block_size > 0);
-        plan.blocks.emplace_back(FileBlock::Range(block_offset, block_offset + block_size - 1));
+        const FileBlock::Range block_range(block_offset, block_offset + block_size - 1);
+        plan.blocks.emplace_back(block_range);
         block_offsets.emplace_back(block_offset);
     }
     DORIS_CHECK(!plan.blocks.empty());
@@ -180,42 +180,42 @@ CachedRemoteFileReader::AsyncReadPlan CachedRemoteFileReader::_build_async_read_
     plan.probe_result.emplace(_cache->probe(_cache_hash, align_left, align_size, cache_context));
     g_cached_remote_reader_probe_total << 1;
     const auto& probe_result = *plan.probe_result;
+    DORIS_CHECK(probe_result.file_blocks.size() == plan.blocks.size());
 
-    const auto overlaps = [](const FileBlock::Range& lhs, const FileBlock::Range& rhs) {
-        return lhs.left <= rhs.right && rhs.left <= lhs.right;
-    };
     for (size_t index = 0; index < plan.blocks.size(); ++index) {
         auto& read_block = plan.blocks[index];
         if (read_block.source == AsyncReadBlock::Source::INFLIGHT) {
             continue;
         }
 
-        bool has_miss = std::any_of(
-                probe_result.gaps.begin(), probe_result.gaps.end(),
-                [&](const FileBlock::Range& gap) { return overlaps(gap, read_block.range); });
-        bool has_downloading = false;
-        bool has_cache_block = false;
-        for (const auto& block : probe_result.holder.file_blocks) {
-            if (!overlaps(block->range(), read_block.range)) {
-                continue;
-            }
-            has_cache_block = true;
-            if (_cache->is_block_deleting(block)) {
-                has_miss = true;
+        const auto& file_block = probe_result.file_blocks[index];
+        bool is_miss = file_block == nullptr;
+        bool is_downloading = false;
+        if (file_block != nullptr) {
+            DORIS_CHECK(file_block->range().left == read_block.range.left);
+            DORIS_CHECK(file_block->range().right == read_block.range.right);
+            if (_cache->is_block_deleting(file_block)) {
+                is_miss = true;
             } else {
-                const FileBlock::State state = block->state();
-                has_miss = has_miss || state == FileBlock::State::EMPTY ||
-                           state == FileBlock::State::SKIP_CACHE;
-                has_downloading = has_downloading || state == FileBlock::State::DOWNLOADING;
+                switch (file_block->state()) {
+                case FileBlock::State::DOWNLOADED:
+                    break;
+                case FileBlock::State::DOWNLOADING:
+                    is_downloading = true;
+                    break;
+                case FileBlock::State::EMPTY:
+                case FileBlock::State::SKIP_CACHE:
+                    is_miss = true;
+                    break;
+                }
             }
         }
-        DORIS_CHECK(has_miss || has_cache_block);
 
-        if (has_miss) {
+        if (is_miss) {
             read_block.submit_write = true;
             ++stats.probe_miss;
             g_cached_remote_reader_probe_miss << 1;
-        } else if (has_downloading) {
+        } else if (is_downloading) {
             read_block.source = AsyncReadBlock::Source::DOWNLOADING;
             ++stats.probe_downloading_hit;
             g_cached_remote_reader_probe_downloading << 1;
@@ -238,10 +238,14 @@ CachedRemoteFileReader::AsyncReadPlan CachedRemoteFileReader::_build_async_read_
 // Copy one block available from inflight memory or cache. DOWNLOADING blocks outside the remote
 // span retain the existing wait semantics. Any race or read failure asks the caller to replace the
 // whole planned request with one remote read instead of incrementally repairing the range.
-bool CachedRemoteFileReader::_materialize_async_block(
-        const AsyncReadPlan& plan, const AsyncReadBlock& read_block, size_t user_offset,
-        Slice result, const CacheContext& cache_context, ReadStatistics& stats,
-        size_t* materialized_bytes, bool* need_self_heal) {
+bool CachedRemoteFileReader::_materialize_async_block(const AsyncReadPlan& plan, size_t block_index,
+                                                      size_t user_offset, Slice result,
+                                                      const CacheContext& cache_context,
+                                                      ReadStatistics& stats,
+                                                      size_t* materialized_bytes,
+                                                      bool* need_self_heal) {
+    DORIS_CHECK(block_index < plan.blocks.size());
+    const auto& read_block = plan.blocks[block_index];
     DORIS_CHECK(read_block.source != AsyncReadBlock::Source::REMOTE);
     DORIS_CHECK(materialized_bytes != nullptr);
     DORIS_CHECK(need_self_heal != nullptr);
@@ -265,64 +269,54 @@ bool CachedRemoteFileReader::_materialize_async_block(
     }
 
     DORIS_CHECK(plan.probe_result.has_value());
-    size_t local_read_bytes = 0;
-    std::vector<FileBlockSPtr> consumed_blocks;
-    for (const auto& block : plan.probe_result->holder.file_blocks) {
-        if (block->range().right < copy_left || block->range().left > copy_right) {
-            continue;
-        }
-        if (_cache->is_block_deleting(block)) {
-            return false;
-        }
+    DORIS_CHECK(block_index < plan.probe_result->file_blocks.size());
+    const auto& file_block = plan.probe_result->file_blocks[block_index];
+    DORIS_CHECK(file_block != nullptr);
+    DORIS_CHECK(file_block->range().left == read_block.range.left);
+    DORIS_CHECK(file_block->range().right == read_block.range.right);
+    if (_cache->is_block_deleting(file_block)) {
+        return false;
+    }
 
-        FileBlock::State state = block->state();
-        if (state == FileBlock::State::DOWNLOADING) {
-            DORIS_CHECK(read_block.source == AsyncReadBlock::Source::DOWNLOADING);
-            {
-                SCOPED_RAW_TIMER(&stats.remote_wait_timer);
-                state = block->wait();
-            }
-            if (state != FileBlock::State::DOWNLOADED) {
-                ++stats.block_wait_timeout;
-                g_cached_remote_reader_block_wait_timeout << 1;
-                return false;
-            }
-            ++stats.block_wait_success;
-            g_cached_remote_reader_block_wait << 1;
+    FileBlock::State state = file_block->state();
+    if (state == FileBlock::State::DOWNLOADING) {
+        DORIS_CHECK(read_block.source == AsyncReadBlock::Source::DOWNLOADING);
+        {
+            SCOPED_RAW_TIMER(&stats.remote_wait_timer);
+            state = file_block->wait();
         }
         if (state != FileBlock::State::DOWNLOADED) {
+            ++stats.block_wait_timeout;
+            g_cached_remote_reader_block_wait_timeout << 1;
             return false;
         }
+        ++stats.block_wait_success;
+        g_cached_remote_reader_block_wait << 1;
+    }
+    if (state != FileBlock::State::DOWNLOADED) {
+        return false;
+    }
 
-        const size_t read_left = std::max(block->range().left, copy_left);
-        const size_t read_right = std::min(block->range().right, copy_right);
-        const size_t read_size = read_right - read_left + 1;
-        Status status;
-        {
-            SCOPED_RAW_TIMER(&stats.local_read_timer);
-            status = block->read(Slice(result.data + (read_left - user_offset), read_size),
-                                 read_left - block->range().left);
-        }
-        if (!status.ok()) {
-            if (status.is<ErrorCode::NOT_FOUND>()) {
-                *need_self_heal = true;
-                g_read_cache_self_heal_on_not_found << 1;
-            }
-            LOG_EVERY_N(WARNING, 100)
-                    << "Read probed file cache block failed, falling back to remote. path="
-                    << path().native() << ", hash=" << _cache_hash.to_string()
-                    << ", offset=" << block->offset() << ", status=" << status;
-            return false;
+    Status status;
+    {
+        SCOPED_RAW_TIMER(&stats.local_read_timer);
+        status = file_block->read(Slice(result.data + (copy_left - user_offset), copy_size),
+                                  copy_left - file_block->range().left);
+    }
+    if (!status.ok()) {
+        if (status.is<ErrorCode::NOT_FOUND>()) {
+            *need_self_heal = true;
+            g_read_cache_self_heal_on_not_found << 1;
         }
-        local_read_bytes += read_size;
-        consumed_blocks.emplace_back(block);
+        LOG_EVERY_N(WARNING, 100)
+                << "Read probed file cache block failed, falling back to remote. path="
+                << path().native() << ", hash=" << _cache_hash.to_string()
+                << ", offset=" << file_block->offset() << ", status=" << status;
+        return false;
     }
-    DORIS_CHECK(local_read_bytes == copy_size);
 
-    for (const auto& block : consumed_blocks) {
-        _cache->touch_probe_block_if_cached(block, cache_context);
-    }
-    *materialized_bytes += local_read_bytes;
+    _cache->touch_probe_block_if_cached(file_block, cache_context);
+    *materialized_bytes += copy_size;
     return true;
 }
 
@@ -339,9 +333,8 @@ bool CachedRemoteFileReader::_materialize_async_cached_sides(
     size_t materialized_bytes = 0;
     const auto materialize_range = [&](size_t begin, size_t end) {
         for (size_t index = begin; index < end; ++index) {
-            if (!_materialize_async_block(plan, plan.blocks[index], user_offset, result,
-                                          cache_context, stats, &materialized_bytes,
-                                          need_self_heal)) {
+            if (!_materialize_async_block(plan, index, user_offset, result, cache_context, stats,
+                                          &materialized_bytes, need_self_heal)) {
                 return false;
             }
         }