blob: 9e8ae44954433f130aa52ea32ee6d9e44da78d28 [file]
/*
* 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/io/cache/lru_cache.h"
#include <atomic>
#include <cstdint>
#include <cstring>
#include <memory>
#include <string>
#include <thread>
#include <vector>
#include "gtest/gtest.h"
#include "paimon/common/io/cache/cache.h"
#include "paimon/common/io/cache/cache_key.h"
#include "paimon/common/memory/memory_segment.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
class LruCacheTest : public ::testing::Test {
public:
void SetUp() override {
pool_ = GetDefaultPool();
}
std::shared_ptr<CacheKey> MakeKey(int64_t position, bool is_index = false) const {
return CacheKey::ForPosition("test_file", position, 64, is_index);
}
std::shared_ptr<CacheValue> MakeValue(int32_t size, char fill_byte = 0,
CacheCallback callback = {}) const {
auto segment = MemorySegment::AllocateHeapMemory(size, pool_.get());
std::memset(segment.MutableData(), fill_byte, size);
return std::make_shared<CacheValue>(segment, std::move(callback));
}
auto MakeSupplier(int32_t size, char fill_byte = 0, CacheCallback callback = {}) const {
return [this, size, fill_byte, callback = std::move(callback)](
const std::shared_ptr<CacheKey>&) -> Result<std::shared_ptr<CacheValue>> {
return MakeValue(size, fill_byte, callback);
};
}
private:
std::shared_ptr<MemoryPool> pool_;
};
/// Verifies basic Get with supplier: cache miss loads via supplier, cache hit returns existing.
TEST_F(LruCacheTest, TestGetCacheHitAndMiss) {
LruCache cache(1024);
auto key = MakeKey(0);
int32_t supplier_call_count = 0;
auto supplier = [&](const std::shared_ptr<CacheKey>&) -> Result<std::shared_ptr<CacheValue>> {
supplier_call_count++;
return MakeValue(64, 'A');
};
// First Get: cache miss, supplier should be called
ASSERT_OK_AND_ASSIGN(auto value1, cache.Get(key, supplier));
ASSERT_EQ(supplier_call_count, 1);
ASSERT_EQ(value1->GetSegment().Size(), 64);
ASSERT_EQ(value1->GetSegment().Get(0), 'A');
ASSERT_EQ(cache.Size(), 1);
// Second Get with same key: cache hit, supplier should NOT be called
ASSERT_OK_AND_ASSIGN(auto value2, cache.Get(key, supplier));
ASSERT_EQ(supplier_call_count, 1);
ASSERT_EQ(value2->GetSegment().Get(0), 'A');
ASSERT_EQ(cache.Size(), 1);
ASSERT_EQ(cache.GetMaxWeight(), 1024);
}
/// Verifies Put inserts new entries and updates existing ones.
TEST_F(LruCacheTest, TestPutInsertAndUpdate) {
LruCache cache(1024);
auto key = MakeKey(0);
auto value_a = MakeValue(64, 'A');
auto value_b = MakeValue(128, 'B');
// Insert
ASSERT_OK(cache.Put(key, value_a));
ASSERT_EQ(cache.Size(), 1);
ASSERT_EQ(cache.GetCurrentWeight(), 64);
// Update with larger value
ASSERT_OK(cache.Put(key, value_b));
ASSERT_EQ(cache.Size(), 1);
ASSERT_EQ(cache.GetCurrentWeight(), 128);
// Verify the updated value is returned
ASSERT_OK_AND_ASSIGN(auto result, cache.Get(key, MakeSupplier(0)));
ASSERT_EQ(result->GetSegment().Size(), 128);
ASSERT_EQ(result->GetSegment().Get(0), 'B');
}
/// Verifies weight-based eviction: when total weight exceeds max, LRU entries are evicted.
TEST_F(LruCacheTest, TestWeightBasedEviction) {
// Cache can hold at most 200 bytes
LruCache cache(200);
auto key0 = MakeKey(0);
auto key1 = MakeKey(1);
auto key2 = MakeKey(2);
// Insert 2 entries of 100 bytes each (total 200, at capacity)
ASSERT_OK(cache.Put(key0, MakeValue(100, 'A')));
ASSERT_OK(cache.Put(key1, MakeValue(100, 'B')));
ASSERT_EQ(cache.Size(), 2);
ASSERT_EQ(cache.GetCurrentWeight(), 200);
// Insert a 3rd entry: should evict key0 (LRU, inserted first)
ASSERT_OK(cache.Put(key2, MakeValue(100, 'C')));
ASSERT_EQ(cache.Size(), 2);
ASSERT_EQ(cache.GetCurrentWeight(), 200);
// key0 should be evicted, Get should call supplier
int32_t supplier_called = 0;
auto supplier = [&](const std::shared_ptr<CacheKey>&) -> Result<std::shared_ptr<CacheValue>> {
supplier_called++;
return MakeValue(100, 'X');
};
ASSERT_OK_AND_ASSIGN(auto result, cache.Get(key0, supplier));
ASSERT_EQ(supplier_called, 1);
ASSERT_EQ(result->GetSegment().Get(0), 'X');
}
/// Verifies that eviction invokes the CacheCallback on evicted entries.
TEST_F(LruCacheTest, TestEvictionCallback) {
LruCache cache(200);
std::vector<int64_t> evicted_positions;
auto make_callback = [&evicted_positions](int64_t position) -> CacheCallback {
return [&evicted_positions, position](const std::shared_ptr<CacheKey>&) {
evicted_positions.push_back(position);
};
};
auto key0 = MakeKey(0);
auto key1 = MakeKey(1);
auto key2 = MakeKey(2);
ASSERT_OK(cache.Put(key0, MakeValue(100, 'A', make_callback(0))));
ASSERT_OK(cache.Put(key1, MakeValue(100, 'B', make_callback(1))));
ASSERT_TRUE(evicted_positions.empty());
// Insert key2: should evict key0 and trigger its callback
ASSERT_OK(cache.Put(key2, MakeValue(100, 'C', make_callback(2))));
ASSERT_EQ(evicted_positions.size(), 1);
ASSERT_EQ(evicted_positions[0], 0);
}
/// Verifies LRU ordering: accessing an entry moves it to the front, preventing eviction.
TEST_F(LruCacheTest, TestLruOrdering) {
LruCache cache(200);
auto key0 = MakeKey(0);
auto key1 = MakeKey(1);
auto key2 = MakeKey(2);
ASSERT_OK(cache.Put(key0, MakeValue(100, 'A')));
ASSERT_OK(cache.Put(key1, MakeValue(100, 'B')));
// Access key0 via Get to move it to front (most recently used)
ASSERT_OK_AND_ASSIGN(auto val, cache.Get(key0, MakeSupplier(0)));
ASSERT_EQ(val->GetSegment().Get(0), 'A');
// Insert key2: should evict key1 (now LRU), NOT key0
std::vector<int64_t> evicted;
auto callback0 = [&evicted](const std::shared_ptr<CacheKey>&) { evicted.push_back(0); };
auto callback1 = [&evicted](const std::shared_ptr<CacheKey>&) { evicted.push_back(1); };
// Re-insert with callbacks to track eviction
cache.InvalidateAll();
ASSERT_OK(cache.Put(key0, MakeValue(100, 'A', callback0)));
ASSERT_OK(cache.Put(key1, MakeValue(100, 'B', callback1)));
// Access key0 to move it to front
ASSERT_OK_AND_ASSIGN(val, cache.Get(key0, MakeSupplier(0)));
// Insert key2: key1 should be evicted (it's at the back)
ASSERT_OK(cache.Put(key2, MakeValue(100, 'C')));
ASSERT_EQ(evicted.size(), 1);
ASSERT_EQ(evicted[0], 1);
ASSERT_EQ(cache.Size(), 2);
}
/// Verifies Invalidate removes a specific entry and adjusts weight.
TEST_F(LruCacheTest, TestInvalidate) {
LruCache cache(1024);
auto key0 = MakeKey(0);
auto key1 = MakeKey(1);
std::vector<int64_t> evicted;
auto callback0 = [&evicted](const std::shared_ptr<CacheKey>&) { evicted.push_back(0); };
auto callback1 = [&evicted](const std::shared_ptr<CacheKey>&) { evicted.push_back(1); };
ASSERT_OK(cache.Put(key0, MakeValue(100, 'A', callback0)));
ASSERT_OK(cache.Put(key1, MakeValue(200, 'B', callback1)));
ASSERT_EQ(cache.Size(), 2);
ASSERT_EQ(cache.GetCurrentWeight(), 300);
ASSERT_TRUE(evicted.empty());
// Invalidate key0
cache.Invalidate(key0);
ASSERT_EQ(cache.Size(), 1);
ASSERT_EQ(cache.GetCurrentWeight(), 200);
ASSERT_EQ(evicted, std::vector<int64_t>({0}));
// Invalidating a non-existent key is a no-op
cache.Invalidate(MakeKey(999));
ASSERT_EQ(cache.Size(), 1);
ASSERT_EQ(cache.GetCurrentWeight(), 200);
ASSERT_EQ(evicted, std::vector<int64_t>({0}));
}
/// Verifies InvalidateAll clears all entries and resets weight.
TEST_F(LruCacheTest, TestInvalidateAll) {
LruCache cache(1024);
std::vector<int64_t> evicted;
for (int32_t i = 0; i < 5; i++) {
auto callback = [&evicted, id = i](const std::shared_ptr<CacheKey>&) {
evicted.push_back(id);
};
ASSERT_OK(cache.Put(MakeKey(i), MakeValue(50, 'A', callback)));
}
ASSERT_EQ(cache.Size(), 5);
ASSERT_EQ(cache.GetCurrentWeight(), 250);
ASSERT_TRUE(evicted.empty());
cache.InvalidateAll();
ASSERT_EQ(cache.Size(), 0);
ASSERT_EQ(cache.GetCurrentWeight(), 0);
ASSERT_EQ(evicted, std::vector<int64_t>({0, 1, 2, 3, 4}));
}
/// Verifies weight tracking is accurate across Put, Update, Invalidate, and Eviction.
TEST_F(LruCacheTest, TestWeightTracking) {
LruCache cache(500);
auto key0 = MakeKey(0);
auto key1 = MakeKey(1);
// Put 100 bytes
ASSERT_OK(cache.Put(key0, MakeValue(100)));
ASSERT_EQ(cache.GetCurrentWeight(), 100);
// Put 200 bytes
ASSERT_OK(cache.Put(key1, MakeValue(200)));
ASSERT_EQ(cache.GetCurrentWeight(), 300);
// Update key0 from 100 to 150 bytes
ASSERT_OK(cache.Put(key0, MakeValue(150)));
ASSERT_EQ(cache.GetCurrentWeight(), 350);
// Invalidate key1 (200 bytes)
cache.Invalidate(key1);
ASSERT_EQ(cache.GetCurrentWeight(), 150);
// Put 200 bytes
ASSERT_OK(cache.Put(MakeKey(2), MakeValue(200)));
ASSERT_EQ(cache.GetCurrentWeight(), 350);
// Add: total would be 550 > 500, should evict key0 (150 bytes)
ASSERT_OK(cache.Put(MakeKey(3), MakeValue(200)));
ASSERT_EQ(cache.GetCurrentWeight(), 400);
ASSERT_EQ(cache.Size(), 2);
}
/// Verifies that Get via supplier correctly handles the case where the supplier returns an error.
TEST_F(LruCacheTest, TestSupplierError) {
LruCache cache(1024);
auto key = MakeKey(0);
auto error_supplier =
[](const std::shared_ptr<CacheKey>&) -> Result<std::shared_ptr<CacheValue>> {
return Status::IOError("simulated read failure");
};
// Get should propagate the supplier error
ASSERT_NOK_WITH_MSG(cache.Get(key, error_supplier), "simulated read failure");
// Cache should remain empty after failed supplier
ASSERT_EQ(cache.Size(), 0);
ASSERT_EQ(cache.GetCurrentWeight(), 0);
}
/// Verifies thread safety: concurrent Get/Put operations should not corrupt internal state.
TEST_F(LruCacheTest, TestConcurrentAccess) {
LruCache cache(10000);
constexpr int32_t num_threads = 8;
constexpr int32_t ops_per_thread = 100;
std::atomic<int32_t> supplier_calls{0};
auto supplier = [&](const std::shared_ptr<CacheKey>&) -> Result<std::shared_ptr<CacheValue>> {
supplier_calls++;
return MakeValue(10, 'X');
};
std::vector<std::thread> threads;
for (int32_t t = 0; t < num_threads; t++) {
threads.emplace_back([&, t]() {
for (int32_t i = 0; i < ops_per_thread; i++) {
auto key = MakeKey(t * ops_per_thread + i);
auto result = cache.Get(key, supplier);
ASSERT_TRUE(result.ok());
}
});
}
for (auto& thread : threads) {
thread.join();
}
// All entries should be in cache (capacity is large enough)
ASSERT_EQ(cache.Size(), num_threads * ops_per_thread);
ASSERT_EQ(supplier_calls.load(), num_threads * ops_per_thread);
}
/// Verifies that Put moves an existing entry to the front of the LRU list.
TEST_F(LruCacheTest, TestPutMovesToFront) {
LruCache cache(200);
auto key0 = MakeKey(0);
auto key1 = MakeKey(1);
auto key2 = MakeKey(2);
std::vector<int64_t> evicted;
auto make_callback = [&evicted](int64_t pos) -> CacheCallback {
return [&evicted, pos](const std::shared_ptr<CacheKey>&) { evicted.push_back(pos); };
};
ASSERT_OK(cache.Put(key0, MakeValue(100, 'A', make_callback(0))));
ASSERT_OK(cache.Put(key1, MakeValue(100, 'B', make_callback(1))));
// Get key0 (should move it to front)
ASSERT_OK(cache.Get(key0, /*supplier=*/nullptr));
// Insert key2: should evict key1 (now at back), not key0
ASSERT_OK(cache.Put(key2, MakeValue(100, 'C', make_callback(2))));
ASSERT_EQ(evicted.size(), 1);
ASSERT_EQ(evicted[0], 1);
}
/// Verifies that multiple evictions happen when a single large entry is inserted.
TEST_F(LruCacheTest, TestMultipleEvictions) {
LruCache cache(300);
std::vector<int64_t> evicted;
auto make_callback = [&evicted](int64_t pos) -> CacheCallback {
return [&evicted, pos](const std::shared_ptr<CacheKey>&) { evicted.push_back(pos); };
};
// Insert 3 entries of 100 bytes each
ASSERT_OK(cache.Put(MakeKey(0), MakeValue(100, 'A', make_callback(0))));
ASSERT_OK(cache.Put(MakeKey(1), MakeValue(100, 'B', make_callback(1))));
ASSERT_OK(cache.Put(MakeKey(2), MakeValue(100, 'C', make_callback(2))));
ASSERT_EQ(cache.Size(), 3);
ASSERT_EQ(cache.GetCurrentWeight(), 300);
// Insert a 250-byte entry: should evict key0, key1 and key2.
ASSERT_OK(cache.Put(MakeKey(3), MakeValue(250, 'D')));
ASSERT_EQ(cache.Size(), 1);
ASSERT_EQ(cache.GetCurrentWeight(), 250);
ASSERT_EQ(evicted, std::vector<int64_t>({0, 1, 2}));
}
} // namespace paimon::test