src/paimon/common/sst/block_cache_test.cpp - paimon-cpp - Git at Google

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

 #include "paimon/common/sst/block_cache.h"

 #include <cstdint>
 #include <cstring>
 #include <memory>
 #include <string>

 #include "gtest/gtest.h"
 #include "paimon/common/io/cache/cache_manager.h"
 #include "paimon/common/io/cache/lru_cache.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/testing/utils/testharness.h"
 namespace paimon::test {

 class BlockCacheTest : public ::testing::Test {
  public:
     void SetUp() override {
         dir_ = UniqueTestDirectory::Create();
         fs_ = dir_->GetFileSystem();
         pool_ = GetDefaultPool();
     }

     void TearDown() override {}

     Status WriteTestFile(const std::string& path, int32_t num_blocks, int32_t block_size) const {
         PAIMON_ASSIGN_OR_RAISE(auto out, fs_->Create(path, false));
         for (int32_t i = 0; i < num_blocks; i++) {
             auto segment = MemorySegment::AllocateHeapMemory(block_size, pool_.get());
             std::memset(segment.MutableData(), i & 0xFF, block_size);
             PAIMON_RETURN_NOT_OK(out->Write(segment.MutableData(), block_size));
         }
         PAIMON_RETURN_NOT_OK(out->Flush());
         PAIMON_RETURN_NOT_OK(out->Close());
         return Status::OK();
     }

     Result<MemorySegment> GetBlock(int32_t block_id, int32_t block_size, BlockCache* block_cache,
                                    bool is_index = false) const {
         return block_cache->GetBlock(/*position=*/block_id * block_size, /*length=*/block_size,
                                      is_index,
                                      /*decompress_func=*/nullptr);
     }

     bool ContainsBlock(int32_t block_id, int32_t block_size, BlockCache* block_cache,
                        bool is_index = false) const {
         return block_cache->ContainsBlock(/*position=*/block_id * block_size, /*length=*/block_size,
                                           is_index);
     }

  private:
     std::unique_ptr<UniqueTestDirectory> dir_;
     std::shared_ptr<FileSystem> fs_;
     std::shared_ptr<MemoryPool> pool_;
 };

 /// Verifies that the first GetBlock call reads from IO and subsequent calls return from the
 /// local blocks_ cache without re-reading.
 TEST_F(BlockCacheTest, TestBasicCacheHit) {
     const int32_t block_size = 64;
     const int32_t num_blocks = 4;
     auto file_path = dir_->Str() + "/basic_hit.data";
     ASSERT_OK(WriteTestFile(file_path, num_blocks, block_size));

     auto cache_manager = std::make_shared<CacheManager>(block_size * num_blocks * 2, 0.0);
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
     BlockCache block_cache(file_path, in, cache_manager, pool_);

     // Initially blocks_ is empty
     ASSERT_EQ(block_cache.BlocksSize(), 0);

     // First access: populates both blocks_ and LRU
     ASSERT_OK_AND_ASSIGN(auto seg1, GetBlock(0, block_size, &block_cache));
     ASSERT_EQ(seg1.Size(), block_size);
     ASSERT_EQ(seg1.Get(0), static_cast<char>(0));
     ASSERT_EQ(block_cache.BlocksSize(), 1);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 1);

     // Second access: returns from blocks_, no new LRU entry
     ASSERT_OK_AND_ASSIGN(auto seg2, GetBlock(0, block_size, &block_cache));
     ASSERT_EQ(seg2.Size(), block_size);
     ASSERT_EQ(block_cache.BlocksSize(), 1);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 1);

     // Load a different block
     ASSERT_OK_AND_ASSIGN(auto seg3, GetBlock(1, block_size, &block_cache));
     ASSERT_EQ(seg3.Get(0), static_cast<char>(1));
     ASSERT_EQ(block_cache.BlocksSize(), 2);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 2);
 }

 /// Verifies that when LRU evicts an entry due to capacity pressure, the eviction callback
 /// removes the corresponding entry from BlockCache's blocks_ map.
 TEST_F(BlockCacheTest, TestLruEvictionSyncsWithBlocks) {
     const int32_t block_size = 100;
     auto file_path = dir_->Str() + "/eviction.data";
     ASSERT_OK(WriteTestFile(file_path, 5, block_size));

     // Cache can hold at most 2 blocks (200 bytes)
     auto cache_manager = std::make_shared<CacheManager>(block_size * 2, 0.0);
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
     BlockCache block_cache(file_path, in, cache_manager, pool_);

     // Load block 0 at position 0
     ASSERT_OK_AND_ASSIGN(auto seg0, GetBlock(0, block_size, &block_cache));
     ASSERT_EQ(seg0.Get(0), static_cast<char>(0));
     ASSERT_EQ(block_cache.BlocksSize(), 1);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));

     // Load block 1 at position block_size
     ASSERT_OK_AND_ASSIGN(auto seg1, GetBlock(1, block_size, &block_cache));
     ASSERT_EQ(seg1.Get(0), static_cast<char>(1));
     ASSERT_EQ(block_cache.BlocksSize(), 2);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));

     // Load block 2: evicts block 0 (LRU) from both LRU and blocks_
     ASSERT_OK_AND_ASSIGN(auto seg2, GetBlock(2, block_size, &block_cache));
     ASSERT_EQ(seg2.Get(0), static_cast<char>(2));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 2);
     ASSERT_EQ(block_cache.BlocksSize(), 2);
     // block 0 should be evicted from blocks_
     ASSERT_FALSE(ContainsBlock(0, block_size, &block_cache));
     // block 1 and block 2 should remain
     ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));
     ASSERT_TRUE(ContainsBlock(2, block_size, &block_cache));

     // Re-access block 0: triggers fresh IO read, evicts block 1 (now LRU)
     ASSERT_OK_AND_ASSIGN(auto seg0_reloaded, GetBlock(0, block_size, &block_cache));
     ASSERT_EQ(seg0_reloaded.Get(0), static_cast<char>(0));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 2);
     ASSERT_EQ(block_cache.BlocksSize(), 2);
     // block 1 should now be evicted
     ASSERT_FALSE(ContainsBlock(1, block_size, &block_cache));
     // block 0 and block 2 should remain
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_TRUE(ContainsBlock(2, block_size, &block_cache));
 }

 /// Verifies that Close() invalidates all entries from both blocks_ and the LRU cache.
 TEST_F(BlockCacheTest, TestClose) {
     const int32_t block_size = 64;
     auto file_path = dir_->Str() + "/close.data";
     ASSERT_OK(WriteTestFile(file_path, 3, block_size));

     auto cache_manager = std::make_shared<CacheManager>(block_size * 10, 0.0);
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
     BlockCache block_cache(file_path, in, cache_manager, pool_);

     // Load 3 blocks and verify blocks_ keys
     for (int32_t i = 0; i < 3; i++) {
         ASSERT_OK_AND_ASSIGN(auto seg, GetBlock(i, block_size, &block_cache));
         ASSERT_EQ(seg.Get(0), static_cast<char>(i));
     }
     ASSERT_EQ(block_cache.BlocksSize(), 3);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));
     ASSERT_TRUE(ContainsBlock(2, block_size, &block_cache));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 3);

     block_cache.Close();

     // After Close, both blocks_ and LRU should be empty
     ASSERT_EQ(block_cache.BlocksSize(), 0);
     ASSERT_FALSE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_FALSE(ContainsBlock(1, block_size, &block_cache));
     ASSERT_FALSE(ContainsBlock(2, block_size, &block_cache));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 0);
 }

 /// Verifies that two BlockCache instances sharing the same CacheManager have independent blocks_
 /// maps, but eviction in the shared LRU only affects the owning BlockCache's blocks_.
 TEST_F(BlockCacheTest, TestSharedCacheManagerEvictionIsolation) {
     const int32_t block_size = 100;
     auto file_path_a = dir_->Str() + "/file_a.data";
     auto file_path_b = dir_->Str() + "/file_b.data";
     ASSERT_OK(WriteTestFile(file_path_a, 3, block_size));
     ASSERT_OK(WriteTestFile(file_path_b, 3, block_size));

     // Shared cache can hold 3 blocks total
     auto cache_manager = std::make_shared<CacheManager>(block_size * 3, 0.0);

     ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in_a, fs_->Open(file_path_a));
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in_b, fs_->Open(file_path_b));
     BlockCache cache_a(file_path_a, in_a, cache_manager, pool_);
     BlockCache cache_b(file_path_b, in_b, cache_manager, pool_);

     // Load 2 blocks from file_a
     ASSERT_OK_AND_ASSIGN(auto seg_a0, GetBlock(0, block_size, &cache_a));
     ASSERT_OK_AND_ASSIGN(auto seg_a1, GetBlock(1, block_size, &cache_a));
     ASSERT_EQ(cache_a.BlocksSize(), 2);
     ASSERT_TRUE(ContainsBlock(0, block_size, &cache_a));
     ASSERT_TRUE(ContainsBlock(1, block_size, &cache_a));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 2);

     // Load 1 block from file_b (total 3, at capacity)
     ASSERT_OK_AND_ASSIGN(auto seg_b0, GetBlock(0, block_size, &cache_b));
     ASSERT_EQ(cache_b.BlocksSize(), 1);
     ASSERT_TRUE(ContainsBlock(0, block_size, &cache_b));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 3);

     // Load another block from file_b: should evict file_a's block 0 (the LRU entry)
     ASSERT_OK_AND_ASSIGN(auto seg_b1, GetBlock(1, block_size, &cache_b));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 3);

     // cache_b should have 2 entries
     ASSERT_EQ(cache_b.BlocksSize(), 2);
     ASSERT_TRUE(ContainsBlock(0, block_size, &cache_b));
     ASSERT_TRUE(ContainsBlock(1, block_size, &cache_b));

     // cache_a's block 0 was evicted by LRU callback, only block 1 remains
     ASSERT_EQ(cache_a.BlocksSize(), 1);
     ASSERT_FALSE(ContainsBlock(0, block_size, &cache_a));
     ASSERT_TRUE(ContainsBlock(1, block_size, &cache_a));

     // Re-access file_a's block 0: triggers fresh IO read, evicts file_a's block 1 (now LRU)
     ASSERT_OK_AND_ASSIGN(auto seg_a0_reloaded, GetBlock(0, block_size, &cache_a));
     ASSERT_EQ(seg_a0_reloaded.Get(0), static_cast<char>(0));
     ASSERT_EQ(cache_a.BlocksSize(), 1);
     ASSERT_TRUE(ContainsBlock(0, block_size, &cache_a));
     ASSERT_FALSE(ContainsBlock(1, block_size, &cache_a));

     // cache_b should be unaffected
     ASSERT_EQ(cache_b.BlocksSize(), 2);
     ASSERT_TRUE(ContainsBlock(0, block_size, &cache_b));
     ASSERT_TRUE(ContainsBlock(1, block_size, &cache_b));

     cache_a.Close();
     cache_b.Close();
 }

 /// Verifies the REFRESH_COUNT mechanism interacts correctly with LRU eviction ordering.
 /// After refreshing a block (re-inserting into LRU front), it should not be the first to be
 /// evicted when capacity pressure occurs.
 TEST_F(BlockCacheTest, TestRefreshPreventsEviction) {
     const int32_t block_size = 100;
     auto file_path = dir_->Str() + "/refresh_eviction.data";
     ASSERT_OK(WriteTestFile(file_path, 4, block_size));

     // Cache can hold 2 blocks
     auto cache_manager = std::make_shared<CacheManager>(block_size * 2, 0.0);
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
     BlockCache block_cache(file_path, in, cache_manager, pool_);

     // Load block 0 and block 1
     ASSERT_OK_AND_ASSIGN(auto seg0, GetBlock(0, block_size, &block_cache));
     ASSERT_OK_AND_ASSIGN(auto seg1, GetBlock(1, block_size, &block_cache));
     ASSERT_EQ(block_cache.BlocksSize(), 2);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 2);

     // Access block 0 REFRESH_COUNT times to trigger a refresh (moves it to LRU front)
     for (int32_t i = 1; i < CacheManager::REFRESH_COUNT; i++) {
         ASSERT_OK_AND_ASSIGN(seg0, GetBlock(0, block_size, &block_cache));
     }
     // This 11th access triggers refresh, moving block 0 to LRU front
     ASSERT_OK_AND_ASSIGN(seg0, GetBlock(0, block_size, &block_cache));

     // blocks_ should still have both entries after refresh
     ASSERT_EQ(block_cache.BlocksSize(), 2);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));

     // Load block 2: should evict block 1 (not block 0, since block 0 was just refreshed)
     ASSERT_OK_AND_ASSIGN(auto seg2, GetBlock(2, block_size, &block_cache));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 2);
     ASSERT_EQ(block_cache.BlocksSize(), 2);
     // block 0 should still be in blocks_ (was refreshed to LRU front)
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
     // block 1 should be evicted from blocks_
     ASSERT_FALSE(ContainsBlock(1, block_size, &block_cache));
     // block 2 should be in blocks_
     ASSERT_TRUE(ContainsBlock(2, block_size, &block_cache));

     // Block 0 should still be accessible from blocks_ cache
     ASSERT_OK_AND_ASSIGN(seg0, GetBlock(0, block_size, &block_cache));
     ASSERT_EQ(seg0.Get(0), static_cast<char>(0));

     // Block 1 was evicted, re-accessing triggers IO read
     ASSERT_OK_AND_ASSIGN(seg1, GetBlock(1, block_size, &block_cache));
     ASSERT_EQ(seg1.Get(0), static_cast<char>(1));
 }

 TEST_F(BlockCacheTest, TestIndexAndDataCache) {
     const int32_t block_size = 100;
     const int32_t num_blocks = 6;
     auto file_path = dir_->Str() + "/index_and_data.data";
     ASSERT_OK(WriteTestFile(file_path, num_blocks, block_size));

     // index cache max weight = 100
     // data cache max weight = 300
     auto cache_manager = std::make_shared<CacheManager>(block_size * 4, 0.25);
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
     BlockCache block_cache(file_path, in, cache_manager, pool_);

     // First access for seg0, index
     ASSERT_OK_AND_ASSIGN(auto seg0, GetBlock(0, block_size, &block_cache, /*is_index=*/true));
     ASSERT_EQ(seg0.Get(0), static_cast<char>(0));
     ASSERT_EQ(block_cache.BlocksSize(), 1);
     ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache, /*is_index=*/true));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 0);
     ASSERT_EQ(cache_manager->IndexCache()->Size(), 1);

     // Second access for seg1, index, seg0 will be evicted
     ASSERT_OK_AND_ASSIGN(auto seg1, GetBlock(1, block_size, &block_cache, /*is_index=*/true));
     ASSERT_EQ(seg1.Get(0), static_cast<char>(1));
     ASSERT_EQ(block_cache.BlocksSize(), 1);
     ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache, /*is_index=*/true));
     ASSERT_FALSE(ContainsBlock(0, block_size, &block_cache, /*is_index=*/true));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 0);
     ASSERT_EQ(cache_manager->IndexCache()->Size(), 1);

     // Fills data cache
     for (int32_t i = 2; i < 5; i++) {
         ASSERT_OK_AND_ASSIGN(auto seg, GetBlock(i, block_size, &block_cache, /*is_index=*/false));
         ASSERT_EQ(seg.Get(0), static_cast<char>(i));
         ASSERT_EQ(block_cache.BlocksSize(), 1 + i - 1);
         ASSERT_TRUE(ContainsBlock(i, block_size, &block_cache, /*is_index=*/false));
         ASSERT_EQ(cache_manager->IndexCache()->Size(), 1);
         ASSERT_EQ(cache_manager->DataCache()->Size(), i - 1);
     }

     ASSERT_OK_AND_ASSIGN(auto seg5, GetBlock(5, block_size, &block_cache, /*is_index=*/false));
     ASSERT_EQ(seg5.Get(0), static_cast<char>(5));
     ASSERT_EQ(block_cache.BlocksSize(), 4);
     ASSERT_TRUE(ContainsBlock(5, block_size, &block_cache, /*is_index=*/false));
     ASSERT_TRUE(ContainsBlock(4, block_size, &block_cache, /*is_index=*/false));
     ASSERT_TRUE(ContainsBlock(3, block_size, &block_cache, /*is_index=*/false));
     ASSERT_FALSE(ContainsBlock(2, block_size, &block_cache, /*is_index=*/false));
     ASSERT_EQ(cache_manager->DataCache()->Size(), 3);
     ASSERT_EQ(cache_manager->IndexCache()->Size(), 1);
 }

 }  // namespace paimon::test
	/*
	* 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.
	*/

	#include "paimon/common/sst/block_cache.h"

	#include <cstdint>
	#include <cstring>
	#include <memory>
	#include <string>

	#include "gtest/gtest.h"
	#include "paimon/common/io/cache/cache_manager.h"
	#include "paimon/common/io/cache/lru_cache.h"
	#include "paimon/memory/memory_pool.h"
	#include "paimon/testing/utils/testharness.h"
	namespace paimon::test {

	class BlockCacheTest : public ::testing::Test {
	public:
	void SetUp() override {
	dir_ = UniqueTestDirectory::Create();
	fs_ = dir_->GetFileSystem();
	pool_ = GetDefaultPool();
	}

	void TearDown() override {}

	Status WriteTestFile(const std::string& path, int32_t num_blocks, int32_t block_size) const {
	PAIMON_ASSIGN_OR_RAISE(auto out, fs_->Create(path, false));
	for (int32_t i = 0; i < num_blocks; i++) {
	auto segment = MemorySegment::AllocateHeapMemory(block_size, pool_.get());
	std::memset(segment.MutableData(), i & 0xFF, block_size);
	PAIMON_RETURN_NOT_OK(out->Write(segment.MutableData(), block_size));
	}
	PAIMON_RETURN_NOT_OK(out->Flush());
	PAIMON_RETURN_NOT_OK(out->Close());
	return Status::OK();
	}

	Result<MemorySegment> GetBlock(int32_t block_id, int32_t block_size, BlockCache* block_cache,
	bool is_index = false) const {
	return block_cache->GetBlock(/position=/block_id * block_size, /length=/block_size,
	is_index,
	/decompress_func=/nullptr);
	}

	bool ContainsBlock(int32_t block_id, int32_t block_size, BlockCache* block_cache,
	bool is_index = false) const {
	return block_cache->ContainsBlock(/position=/block_id * block_size, /length=/block_size,
	is_index);
	}

	private:
	std::unique_ptr<UniqueTestDirectory> dir_;
	std::shared_ptr<FileSystem> fs_;
	std::shared_ptr<MemoryPool> pool_;
	};

	/// Verifies that the first GetBlock call reads from IO and subsequent calls return from the
	/// local blocks_ cache without re-reading.
	TEST_F(BlockCacheTest, TestBasicCacheHit) {
	const int32_t block_size = 64;
	const int32_t num_blocks = 4;
	auto file_path = dir_->Str() + "/basic_hit.data";
	ASSERT_OK(WriteTestFile(file_path, num_blocks, block_size));

	auto cache_manager = std::make_shared<CacheManager>(block_size * num_blocks * 2, 0.0);
	ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
	BlockCache block_cache(file_path, in, cache_manager, pool_);

	// Initially blocks_ is empty
	ASSERT_EQ(block_cache.BlocksSize(), 0);

	// First access: populates both blocks_ and LRU
	ASSERT_OK_AND_ASSIGN(auto seg1, GetBlock(0, block_size, &block_cache));
	ASSERT_EQ(seg1.Size(), block_size);
	ASSERT_EQ(seg1.Get(0), static_cast<char>(0));
	ASSERT_EQ(block_cache.BlocksSize(), 1);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 1);

	// Second access: returns from blocks_, no new LRU entry
	ASSERT_OK_AND_ASSIGN(auto seg2, GetBlock(0, block_size, &block_cache));
	ASSERT_EQ(seg2.Size(), block_size);
	ASSERT_EQ(block_cache.BlocksSize(), 1);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 1);

	// Load a different block
	ASSERT_OK_AND_ASSIGN(auto seg3, GetBlock(1, block_size, &block_cache));
	ASSERT_EQ(seg3.Get(0), static_cast<char>(1));
	ASSERT_EQ(block_cache.BlocksSize(), 2);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 2);
	}

	/// Verifies that when LRU evicts an entry due to capacity pressure, the eviction callback
	/// removes the corresponding entry from BlockCache's blocks_ map.
	TEST_F(BlockCacheTest, TestLruEvictionSyncsWithBlocks) {
	const int32_t block_size = 100;
	auto file_path = dir_->Str() + "/eviction.data";
	ASSERT_OK(WriteTestFile(file_path, 5, block_size));

	// Cache can hold at most 2 blocks (200 bytes)
	auto cache_manager = std::make_shared<CacheManager>(block_size * 2, 0.0);
	ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
	BlockCache block_cache(file_path, in, cache_manager, pool_);

	// Load block 0 at position 0
	ASSERT_OK_AND_ASSIGN(auto seg0, GetBlock(0, block_size, &block_cache));
	ASSERT_EQ(seg0.Get(0), static_cast<char>(0));
	ASSERT_EQ(block_cache.BlocksSize(), 1);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));

	// Load block 1 at position block_size
	ASSERT_OK_AND_ASSIGN(auto seg1, GetBlock(1, block_size, &block_cache));
	ASSERT_EQ(seg1.Get(0), static_cast<char>(1));
	ASSERT_EQ(block_cache.BlocksSize(), 2);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));

	// Load block 2: evicts block 0 (LRU) from both LRU and blocks_
	ASSERT_OK_AND_ASSIGN(auto seg2, GetBlock(2, block_size, &block_cache));
	ASSERT_EQ(seg2.Get(0), static_cast<char>(2));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 2);
	ASSERT_EQ(block_cache.BlocksSize(), 2);
	// block 0 should be evicted from blocks_
	ASSERT_FALSE(ContainsBlock(0, block_size, &block_cache));
	// block 1 and block 2 should remain
	ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));
	ASSERT_TRUE(ContainsBlock(2, block_size, &block_cache));

	// Re-access block 0: triggers fresh IO read, evicts block 1 (now LRU)
	ASSERT_OK_AND_ASSIGN(auto seg0_reloaded, GetBlock(0, block_size, &block_cache));
	ASSERT_EQ(seg0_reloaded.Get(0), static_cast<char>(0));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 2);
	ASSERT_EQ(block_cache.BlocksSize(), 2);
	// block 1 should now be evicted
	ASSERT_FALSE(ContainsBlock(1, block_size, &block_cache));
	// block 0 and block 2 should remain
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_TRUE(ContainsBlock(2, block_size, &block_cache));
	}

	/// Verifies that Close() invalidates all entries from both blocks_ and the LRU cache.
	TEST_F(BlockCacheTest, TestClose) {
	const int32_t block_size = 64;
	auto file_path = dir_->Str() + "/close.data";
	ASSERT_OK(WriteTestFile(file_path, 3, block_size));

	auto cache_manager = std::make_shared<CacheManager>(block_size * 10, 0.0);
	ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
	BlockCache block_cache(file_path, in, cache_manager, pool_);

	// Load 3 blocks and verify blocks_ keys
	for (int32_t i = 0; i < 3; i++) {
	ASSERT_OK_AND_ASSIGN(auto seg, GetBlock(i, block_size, &block_cache));
	ASSERT_EQ(seg.Get(0), static_cast<char>(i));
	}
	ASSERT_EQ(block_cache.BlocksSize(), 3);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));
	ASSERT_TRUE(ContainsBlock(2, block_size, &block_cache));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 3);

	block_cache.Close();

	// After Close, both blocks_ and LRU should be empty
	ASSERT_EQ(block_cache.BlocksSize(), 0);
	ASSERT_FALSE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_FALSE(ContainsBlock(1, block_size, &block_cache));
	ASSERT_FALSE(ContainsBlock(2, block_size, &block_cache));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 0);
	}

	/// Verifies that two BlockCache instances sharing the same CacheManager have independent blocks_
	/// maps, but eviction in the shared LRU only affects the owning BlockCache's blocks_.
	TEST_F(BlockCacheTest, TestSharedCacheManagerEvictionIsolation) {
	const int32_t block_size = 100;
	auto file_path_a = dir_->Str() + "/file_a.data";
	auto file_path_b = dir_->Str() + "/file_b.data";
	ASSERT_OK(WriteTestFile(file_path_a, 3, block_size));
	ASSERT_OK(WriteTestFile(file_path_b, 3, block_size));

	// Shared cache can hold 3 blocks total
	auto cache_manager = std::make_shared<CacheManager>(block_size * 3, 0.0);

	ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in_a, fs_->Open(file_path_a));
	ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in_b, fs_->Open(file_path_b));
	BlockCache cache_a(file_path_a, in_a, cache_manager, pool_);
	BlockCache cache_b(file_path_b, in_b, cache_manager, pool_);

	// Load 2 blocks from file_a
	ASSERT_OK_AND_ASSIGN(auto seg_a0, GetBlock(0, block_size, &cache_a));
	ASSERT_OK_AND_ASSIGN(auto seg_a1, GetBlock(1, block_size, &cache_a));
	ASSERT_EQ(cache_a.BlocksSize(), 2);
	ASSERT_TRUE(ContainsBlock(0, block_size, &cache_a));
	ASSERT_TRUE(ContainsBlock(1, block_size, &cache_a));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 2);

	// Load 1 block from file_b (total 3, at capacity)
	ASSERT_OK_AND_ASSIGN(auto seg_b0, GetBlock(0, block_size, &cache_b));
	ASSERT_EQ(cache_b.BlocksSize(), 1);
	ASSERT_TRUE(ContainsBlock(0, block_size, &cache_b));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 3);

	// Load another block from file_b: should evict file_a's block 0 (the LRU entry)
	ASSERT_OK_AND_ASSIGN(auto seg_b1, GetBlock(1, block_size, &cache_b));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 3);

	// cache_b should have 2 entries
	ASSERT_EQ(cache_b.BlocksSize(), 2);
	ASSERT_TRUE(ContainsBlock(0, block_size, &cache_b));
	ASSERT_TRUE(ContainsBlock(1, block_size, &cache_b));

	// cache_a's block 0 was evicted by LRU callback, only block 1 remains
	ASSERT_EQ(cache_a.BlocksSize(), 1);
	ASSERT_FALSE(ContainsBlock(0, block_size, &cache_a));
	ASSERT_TRUE(ContainsBlock(1, block_size, &cache_a));

	// Re-access file_a's block 0: triggers fresh IO read, evicts file_a's block 1 (now LRU)
	ASSERT_OK_AND_ASSIGN(auto seg_a0_reloaded, GetBlock(0, block_size, &cache_a));
	ASSERT_EQ(seg_a0_reloaded.Get(0), static_cast<char>(0));
	ASSERT_EQ(cache_a.BlocksSize(), 1);
	ASSERT_TRUE(ContainsBlock(0, block_size, &cache_a));
	ASSERT_FALSE(ContainsBlock(1, block_size, &cache_a));

	// cache_b should be unaffected
	ASSERT_EQ(cache_b.BlocksSize(), 2);
	ASSERT_TRUE(ContainsBlock(0, block_size, &cache_b));
	ASSERT_TRUE(ContainsBlock(1, block_size, &cache_b));

	cache_a.Close();
	cache_b.Close();
	}

	/// Verifies the REFRESH_COUNT mechanism interacts correctly with LRU eviction ordering.
	/// After refreshing a block (re-inserting into LRU front), it should not be the first to be
	/// evicted when capacity pressure occurs.
	TEST_F(BlockCacheTest, TestRefreshPreventsEviction) {
	const int32_t block_size = 100;
	auto file_path = dir_->Str() + "/refresh_eviction.data";
	ASSERT_OK(WriteTestFile(file_path, 4, block_size));

	// Cache can hold 2 blocks
	auto cache_manager = std::make_shared<CacheManager>(block_size * 2, 0.0);
	ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
	BlockCache block_cache(file_path, in, cache_manager, pool_);

	// Load block 0 and block 1
	ASSERT_OK_AND_ASSIGN(auto seg0, GetBlock(0, block_size, &block_cache));
	ASSERT_OK_AND_ASSIGN(auto seg1, GetBlock(1, block_size, &block_cache));
	ASSERT_EQ(block_cache.BlocksSize(), 2);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 2);

	// Access block 0 REFRESH_COUNT times to trigger a refresh (moves it to LRU front)
	for (int32_t i = 1; i < CacheManager::REFRESH_COUNT; i++) {
	ASSERT_OK_AND_ASSIGN(seg0, GetBlock(0, block_size, &block_cache));
	}
	// This 11th access triggers refresh, moving block 0 to LRU front
	ASSERT_OK_AND_ASSIGN(seg0, GetBlock(0, block_size, &block_cache));

	// blocks_ should still have both entries after refresh
	ASSERT_EQ(block_cache.BlocksSize(), 2);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache));

	// Load block 2: should evict block 1 (not block 0, since block 0 was just refreshed)
	ASSERT_OK_AND_ASSIGN(auto seg2, GetBlock(2, block_size, &block_cache));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 2);
	ASSERT_EQ(block_cache.BlocksSize(), 2);
	// block 0 should still be in blocks_ (was refreshed to LRU front)
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache));
	// block 1 should be evicted from blocks_
	ASSERT_FALSE(ContainsBlock(1, block_size, &block_cache));
	// block 2 should be in blocks_
	ASSERT_TRUE(ContainsBlock(2, block_size, &block_cache));

	// Block 0 should still be accessible from blocks_ cache
	ASSERT_OK_AND_ASSIGN(seg0, GetBlock(0, block_size, &block_cache));
	ASSERT_EQ(seg0.Get(0), static_cast<char>(0));

	// Block 1 was evicted, re-accessing triggers IO read
	ASSERT_OK_AND_ASSIGN(seg1, GetBlock(1, block_size, &block_cache));
	ASSERT_EQ(seg1.Get(0), static_cast<char>(1));
	}

	TEST_F(BlockCacheTest, TestIndexAndDataCache) {
	const int32_t block_size = 100;
	const int32_t num_blocks = 6;
	auto file_path = dir_->Str() + "/index_and_data.data";
	ASSERT_OK(WriteTestFile(file_path, num_blocks, block_size));

	// index cache max weight = 100
	// data cache max weight = 300
	auto cache_manager = std::make_shared<CacheManager>(block_size * 4, 0.25);
	ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_path));
	BlockCache block_cache(file_path, in, cache_manager, pool_);

	// First access for seg0, index
	ASSERT_OK_AND_ASSIGN(auto seg0, GetBlock(0, block_size, &block_cache, /is_index=/true));
	ASSERT_EQ(seg0.Get(0), static_cast<char>(0));
	ASSERT_EQ(block_cache.BlocksSize(), 1);
	ASSERT_TRUE(ContainsBlock(0, block_size, &block_cache, /is_index=/true));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 0);
	ASSERT_EQ(cache_manager->IndexCache()->Size(), 1);

	// Second access for seg1, index, seg0 will be evicted
	ASSERT_OK_AND_ASSIGN(auto seg1, GetBlock(1, block_size, &block_cache, /is_index=/true));
	ASSERT_EQ(seg1.Get(0), static_cast<char>(1));
	ASSERT_EQ(block_cache.BlocksSize(), 1);
	ASSERT_TRUE(ContainsBlock(1, block_size, &block_cache, /is_index=/true));
	ASSERT_FALSE(ContainsBlock(0, block_size, &block_cache, /is_index=/true));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 0);
	ASSERT_EQ(cache_manager->IndexCache()->Size(), 1);

	// Fills data cache
	for (int32_t i = 2; i < 5; i++) {
	ASSERT_OK_AND_ASSIGN(auto seg, GetBlock(i, block_size, &block_cache, /is_index=/false));
	ASSERT_EQ(seg.Get(0), static_cast<char>(i));
	ASSERT_EQ(block_cache.BlocksSize(), 1 + i - 1);
	ASSERT_TRUE(ContainsBlock(i, block_size, &block_cache, /is_index=/false));
	ASSERT_EQ(cache_manager->IndexCache()->Size(), 1);
	ASSERT_EQ(cache_manager->DataCache()->Size(), i - 1);
	}

	ASSERT_OK_AND_ASSIGN(auto seg5, GetBlock(5, block_size, &block_cache, /is_index=/false));
	ASSERT_EQ(seg5.Get(0), static_cast<char>(5));
	ASSERT_EQ(block_cache.BlocksSize(), 4);
	ASSERT_TRUE(ContainsBlock(5, block_size, &block_cache, /is_index=/false));
	ASSERT_TRUE(ContainsBlock(4, block_size, &block_cache, /is_index=/false));
	ASSERT_TRUE(ContainsBlock(3, block_size, &block_cache, /is_index=/false));
	ASSERT_FALSE(ContainsBlock(2, block_size, &block_cache, /is_index=/false));
	ASSERT_EQ(cache_manager->DataCache()->Size(), 3);
	ASSERT_EQ(cache_manager->IndexCache()->Size(), 1);
	}

	} // namespace paimon::test