[fix](filecache) fix clear_file_cache right after reboot causing file cache size percent overflow

be/src/io/cache/block_file_cache.cpp

@@ -1243,6 +1243,7 @@ void BlockFileCache::reset_range(const UInt128Wrapper& hash, size_t offset, size
                      << " new_size=" << new_size;
         return;
     }
+    DCHECK_EQ(cell->file_block->_block_range.size(), old_size);
     if (cell->queue_iterator) {
         auto& queue = get_queue(cell->file_block->cache_type());
         DCHECK(queue.contains(hash, offset, cache_lock));
@@ -1251,8 +1252,13 @@ void BlockFileCache::reset_range(const UInt128Wrapper& hash, size_t offset, size
                                           cell->file_block->get_hash_value(),
                                           cell->file_block->offset(), new_size);
     }
+    cell->file_block->_block_range.right = cell->file_block->_block_range.left + new_size - 1;
     _cur_cache_size -= old_size;
     _cur_cache_size += new_size;
+    if (cell->file_block->cache_type() == FileCacheType::TTL) {
+        _cur_ttl_size -= old_size;
+        _cur_ttl_size += new_size;
+    }
 }
 
 bool BlockFileCache::try_reserve_from_other_queue_by_time_interval(

be/src/io/cache/block_file_cache.h

@@ -275,7 +275,7 @@ class BlockFileCache {
     void remove_if_cached(const UInt128Wrapper& key);
     void remove_if_cached_async(const UInt128Wrapper& key);
 
-    // Shrink the block size. old_size is always larged than new_size.
+    // Reset the block size and keep FileBlock, LRU queue, and cache counters consistent.
     void reset_range(const UInt128Wrapper&, size_t offset, size_t old_size, size_t new_size,
                      std::lock_guard<std::mutex>& cache_lock);
 

be/src/io/cache/file_block.cpp

@@ -163,8 +163,7 @@ Status FileBlock::finalize() {
     if (_downloaded_size != _block_range.size()) {
         SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
         size_t old_size = _block_range.size();
-        _block_range.right = _block_range.left + _downloaded_size - 1;
-        size_t new_size = _block_range.size();
+        size_t new_size = _downloaded_size;
         DCHECK(new_size < old_size);
         _mgr->reset_range(_key.hash, _block_range.left, old_size, new_size, cache_lock);
     }

be/test/io/cache/block_file_cache_test.cpp

@@ -8151,4 +8151,75 @@ TEST_F(BlockFileCacheTest, add_cell_rejects_oversized_size) {
     }
 }
 
+TEST_F(BlockFileCacheTest, lru_restore_size_mismatch_does_not_underflow_on_clear) {
+    std::string my_cache_path =
+            caches_dir / "lru_restore_size_mismatch_does_not_underflow_on_clear" / "";
+    if (fs::exists(my_cache_path)) {
+        fs::remove_all(my_cache_path);
+    }
+    fs::create_directories(my_cache_path);
+
+    io::FileCacheSettings settings;
+    settings.query_queue_size = 1_mb;
+    settings.query_queue_elements = 30;
+    settings.capacity = 1_mb;
+    settings.max_file_block_size = 1_mb;
+    settings.max_query_cache_size = 1_mb;
+
+    io::BlockFileCache cache(my_cache_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("lru_restore_size_mismatch_does_not_underflow_on_clear");
+    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;
+
+    constexpr size_t old_size = 128_kb;
+    constexpr size_t new_size = 64_kb;
+    constexpr size_t offset = 0;
+
+    {
+        std::lock_guard<std::mutex> cache_lock(cache._mutex);
+        auto* cell = cache.add_cell(hash, ctx, offset, old_size, io::FileBlock::State::DOWNLOADED,
+                                    cache_lock);
+        ASSERT_NE(cell, nullptr);
+        ASSERT_EQ(cell->file_block->range().size(), old_size);
+        ASSERT_EQ(cache._cur_cache_size, old_size);
+
+        auto* storage = dynamic_cast<io::FSFileCacheStorage*>(cache._storage.get());
+        ASSERT_NE(storage, nullptr);
+        ASSERT_TRUE(storage->handle_already_loaded_block(&cache, hash, offset, new_size,
+                                                         /*tablet_id*/ 0, cache_lock));
+        ASSERT_EQ(cell->file_block->range().size(), new_size);
+        ASSERT_EQ(cell->file_block->range().right, offset + new_size - 1);
+        ASSERT_EQ(cache._cur_cache_size, new_size);
+        ASSERT_EQ(cache.get_queue(io::FileCacheType::NORMAL).get_capacity(cache_lock), new_size);
+    }
+
+    cache.clear_file_cache_async();
+
+    {
+        std::lock_guard<std::mutex> cache_lock(cache._mutex);
+        ASSERT_EQ(cache._cur_cache_size, 0);
+        ASSERT_EQ(cache.get_queue(io::FileCacheType::NORMAL).get_capacity(cache_lock), 0);
+        ASSERT_TRUE(cache._files.empty());
+    }
+
+    if (fs::exists(my_cache_path)) {
+        fs::remove_all(my_cache_path);
+    }
+}
+
 } // namespace doris::io