src/paimon/common/io/cache/lru_cache_test.cpp

/*
 * 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/cache/cache.h"
#include "paimon/common/io/cache/cache_key.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/memory/memory_segment.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');
}

TEST_F(LruCacheTest, TestRetainedMemoryAccounting) {
    LruCache cache(192);
    auto segment = MemorySegment::AllocateHeapMemory(64, pool_.get());
    auto value = std::make_shared<CacheValue>(segment, CacheCallback(), 128);
    ASSERT_EQ(64, value->GetSegment().Size());
    ASSERT_EQ(128, value->GetMemoryUsage());
    ASSERT_EQ(64, CacheValue(segment, CacheCallback()).GetMemoryUsage());
    ASSERT_EQ(64, CacheValue(segment, CacheCallback(), 32).GetMemoryUsage());

    std::vector<std::shared_ptr<CacheKey>> evicted;
    auto first = std::make_shared<CacheValue>(
        segment, [&](const std::shared_ptr<CacheKey>& key) { evicted.push_back(key); }, 128);
    auto first_key = MakeKey(0);
    ASSERT_OK(cache.Put(first_key, first));
    ASSERT_EQ(128, cache.GetCurrentWeight());
    ASSERT_OK(cache.Put(MakeKey(1), value));
    ASSERT_EQ(1, cache.Size());
    ASSERT_EQ(128, cache.GetCurrentWeight());
    ASSERT_EQ(std::vector<std::shared_ptr<CacheKey>>{first_key}, evicted);
    cache.InvalidateAll();
    ASSERT_EQ(0, cache.GetCurrentWeight());

    // The payload fits, but its retained allocation exceeds the cache budget.
    LruCache small_cache(100);
    ASSERT_OK_AND_ASSIGN(
        std::shared_ptr<CacheValue> loaded,
        small_cache.Get(MakeKey(0),
                        [&](const std::shared_ptr<CacheKey>&)
                            -> Result<std::shared_ptr<CacheValue>> { return value; }));
    ASSERT_EQ(value, loaded);
    ASSERT_EQ(0, small_cache.Size());
    ASSERT_EQ(0, small_cache.GetCurrentWeight());
}

/// 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);
}

TEST_F(LruCacheTest, TestForKindSetsKeyKind) {
    LruCache cache(1024);

    auto key = CacheKey::ForKind("test_file", 0, 64, CacheKind::MANIFEST);
    bool supplier_seen_kind = false;
    auto supplier =
        [&](const std::shared_ptr<CacheKey>& supplier_key) -> Result<std::shared_ptr<CacheValue>> {
        supplier_seen_kind = supplier_key->GetKind() == CacheKind::MANIFEST;
        return MakeValue(64, 'M');
    };

    ASSERT_OK_AND_ASSIGN(auto value, cache.Get(key, supplier));
    ASSERT_TRUE(supplier_seen_kind);
    ASSERT_EQ(CacheKind::MANIFEST, key->GetKind());
    ASSERT_EQ('M', value->GetSegment().Get(0));

    auto put_key = CacheKey::ForKind("test_file", 1, 64, CacheKind::MANIFEST);
    ASSERT_OK(cache.Put(put_key, MakeValue(64, 'P')));
    ASSERT_EQ(CacheKind::MANIFEST, put_key->GetKind());
}

TEST_F(LruCacheTest, InferredManifestCacheKeysIncludeBucketCountAndSchema) {
    auto key = SnapshotLiveManifestEntriesCacheKey::ForInferred("table", "main", 1, 4, 0);
    auto same = SnapshotLiveManifestEntriesCacheKey::ForInferred("table", "main", 1, 4, 0);
    ASSERT_TRUE(key->Equals(*same));
    ASSERT_EQ(key->HashCode(), same->HashCode());
    auto explicit_key = SnapshotLiveManifestEntriesCacheKey::ForExplicit("table", "main", 1);
    // Schema zero is real inferred metadata, not an explicit-mode placeholder.
    ASSERT_FALSE(key->Equals(*explicit_key));
    ASSERT_FALSE(explicit_key->Equals(*key));
    ASSERT_FALSE(
        key->Equals(*SnapshotLiveManifestEntriesCacheKey::ForInferred("table", "main", 1, 8, 0)));
    ASSERT_FALSE(
        key->Equals(*SnapshotLiveManifestEntriesCacheKey::ForInferred("table", "main", 1, 4, 1)));
}

TEST_F(LruCacheTest, TestExplicitSnapshotLiveManifestEntriesKeys) {
    auto main_key = SnapshotLiveManifestEntriesCacheKey::ForExplicit("table_path", "main", 0);
    auto same_key = SnapshotLiveManifestEntriesCacheKey::ForExplicit("table_path", "main", 0);
    auto branch_key = SnapshotLiveManifestEntriesCacheKey::ForExplicit("table_path", "dev", 0);
    auto table_key =
        SnapshotLiveManifestEntriesCacheKey::ForExplicit("other_table_path", "main", 0);
    auto bucket_key = SnapshotLiveManifestEntriesCacheKey::ForExplicit("table_path", "main", 1);
    auto hash_in_path_key =
        SnapshotLiveManifestEntriesCacheKey::ForExplicit("table#path", "main", 0);
    auto hash_in_branch_key =
        SnapshotLiveManifestEntriesCacheKey::ForExplicit("table", "path#main", 0);

    ASSERT_EQ(CacheKind::SNAPSHOT_LIVE_MANIFEST, main_key->GetKind());
    ASSERT_TRUE(CacheKeyEqual()(main_key, same_key));
    ASSERT_FALSE(CacheKeyEqual()(main_key, branch_key));
    ASSERT_FALSE(CacheKeyEqual()(main_key, table_key));
    ASSERT_FALSE(CacheKeyEqual()(main_key, bucket_key));
    ASSERT_FALSE(CacheKeyEqual()(hash_in_path_key, hash_in_branch_key));
}

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