[fix](filecache) reject oversized block size in add_cell (apache#62878)

When directory_entry::file_size(ec) fails, it returns (uintmax_t)-1.
load_cache_info_into_memory did not check ec, so UINT64_MAX flowed into
add_cell, which only warned and then registered the cell — corrupting
_files, LRUQueue::cache_size, _cur_cache_size, and the LRU dump
(survives restart).

- fs_file_cache_storage: check ec after file_size, skip on error
- add_cell: turn the >1GiB check into a hard reject (return nullptr)
- UT: add_cell_rejects_oversized_size

Author: deardeng

be/src/io/cache/block_file_cache.cpp

@@ -881,9 +881,10 @@ FileBlockCell* BlockFileCache::add_cell(const UInt128Wrapper& hash, const CacheC
                << " tablet_id=" << context.tablet_id;
 
     if (size > 1024 * 1024 * 1024) {
-        LOG(WARNING) << "File block size is too large for a block. size=" << size
+        LOG(WARNING) << "File block size is too large for a block, reject. size=" << size
                      << " hash=" << hash.to_string() << " offset=" << offset
                      << " stack:" << get_stack_trace();
+        return nullptr;
     }
 
     auto& offsets = _files[hash];

be/src/io/cache/block_file_cache.h

@@ -169,6 +169,7 @@ class BlockFileCache {
     friend class CacheLRUDumper;
     friend class LRUQueueRecorder;
     friend struct FileBlockCell;
+    friend class BlockFileCacheTest;
 
 public:
     // hash the file_name to uint128

be/src/io/cache/fs_file_cache_storage.cpp

@@ -695,6 +695,11 @@ void FSFileCacheStorage::load_cache_info_into_memory_from_fs(BlockFileCache* mgr
             context.expiration_time = expiration_time;
             for (; offset_it != std::filesystem::directory_iterator(); ++offset_it) {
                 size_t size = offset_it->file_size(ec);
+                if (ec) [[unlikely]] {
+                    LOG(WARNING) << "skip cache file, file_size failed, file="
+                                 << offset_it->path().native() << " err=" << ec.message();
+                    continue;
+                }
                 size_t offset = 0;
                 bool is_tmp = false;
                 FileCacheType cache_type = FileCacheType::NORMAL;

be/test/io/cache/block_file_cache_test.cpp

@@ -8058,4 +8058,80 @@ TEST_F(BlockFileCacheTest, set_downloaded_empty_block_branch) {
     ASSERT_EQ(block.get_downloader(), 0);
 }
 
+// Reproduces OPENSOURCE-325: when fs_file_cache_storage's directory scan reads
+// a file size via std::filesystem::directory_entry::file_size(ec) and the
+// underlying syscall fails (e.g. file removed concurrently by clear/GC), the
+// returned value is (uintmax_t)-1 == UINT64_MAX. Before the fix, that bogus
+// size flowed into BlockFileCache::add_cell, which only logged a WARNING and
+// then registered the cell, polluting _files, the LRU queue's cache_size, and
+// _cur_cache_size. After the fix, add_cell rejects sizes > 1GiB and returns
+// nullptr without touching any internal state.
+TEST_F(BlockFileCacheTest, add_cell_rejects_oversized_size) {
+    if (fs::exists(cache_base_path)) {
+        fs::remove_all(cache_base_path);
+    }
+    fs::create_directories(cache_base_path);
+
+    io::FileCacheSettings settings;
+    settings.query_queue_size = 30_mb;
+    settings.query_queue_elements = 30;
+    settings.capacity = 30_mb;
+    settings.max_file_block_size = 1_mb;
+    settings.max_query_cache_size = 30;
+
+    io::BlockFileCache cache(cache_base_path, settings);
+    ASSERT_TRUE(cache.initialize());
+    for (int i = 0; i < 100; ++i) {
+        if (cache.get_async_open_success()) {
+            break;
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    ASSERT_TRUE(cache.get_async_open_success());
+
+    auto hash = io::BlockFileCache::hash("opensource_325_key");
+    io::CacheContext ctx;
+    TUniqueId qid;
+    qid.hi = 1;
+    qid.lo = 1;
+    ctx.query_id = qid;
+    ctx.cache_type = io::FileCacheType::NORMAL;
+    ReadStatistics rstats;
+    ctx.stats = &rstats;
+
+    const size_t kBadSize = std::numeric_limits<size_t>::max(); // (uintmax_t)-1
+    const size_t kOffset = 1128267776;                          // matches the JIRA report
+
+    size_t cur_cache_size_before = 0;
+    size_t normal_queue_size_before = 0;
+    {
+        std::lock_guard<std::mutex> cache_lock(cache._mutex);
+        cur_cache_size_before = cache._cur_cache_size;
+        normal_queue_size_before =
+                cache.get_queue(io::FileCacheType::NORMAL).get_capacity(cache_lock);
+    }
+
+    {
+        std::lock_guard<std::mutex> cache_lock(cache._mutex);
+        auto* cell = cache.add_cell(hash, ctx, kOffset, kBadSize, io::FileBlock::State::DOWNLOADED,
+                                    cache_lock);
+        // Defensive guard must reject oversized blocks instead of polluting state.
+        ASSERT_EQ(cell, nullptr);
+    }
+
+    {
+        std::lock_guard<std::mutex> cache_lock(cache._mutex);
+        // _files index untouched: no entry for (hash, kOffset).
+        ASSERT_EQ(cache._files.find(hash), cache._files.end());
+        // Global and per-queue size counters untouched (no UINT64_MAX added).
+        ASSERT_EQ(cache._cur_cache_size, cur_cache_size_before);
+        ASSERT_EQ(cache.get_queue(io::FileCacheType::NORMAL).get_capacity(cache_lock),
+                  normal_queue_size_before);
+    }
+
+    if (fs::exists(cache_base_path)) {
+        fs::remove_all(cache_base_path);
+    }
+}
+
 } // namespace doris::io