src/paimon/common/utils/read_ahead_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/utils/read_ahead_cache.h"

#include <chrono>
#include <fstream>
#include <limits>
#include <memory>
#include <optional>
#include <thread>
#include <vector>

#include "gtest/gtest.h"
#include "paimon/common/factories/io_hook.h"
#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/utils/scope_guard.h"
#include "paimon/fs/file_system.h"
#include "paimon/fs/file_system_factory.h"
#include "paimon/testing/utils/gated_async_input_stream.h"
#include "paimon/testing/utils/testharness.h"

namespace paimon::test {

// The range limits the tests exercise. The defaults are sized for real data
// files, which the small test files would never reach. The ranges registered
// mid-read are cut at the same limit as the ones registered up front, so a test
// that does not care about the late granularity sees one range per registration.
CacheConfig TestCacheConfig(uint64_t range_size_limit, uint64_t hole_size_limit,
                            uint64_t pre_buffer_limit) {
    CacheConfig config;
    config.SetRangeSizeLimit(range_size_limit);
    config.SetHoleSizeLimit(hole_size_limit);
    config.SetPreBufferLimit(pre_buffer_limit);
    return config;
}

// A pool of its own for every cache, so that a buffer outliving the pool it was
// allocated from shows up instead of being covered by the global pool, which
// never goes away.
std::shared_ptr<MemoryPool> TestPool() {
    return std::shared_ptr<MemoryPool>(GetMemoryPool());
}

// Write the given content into a fresh directory and open it for reading. The
// directory is returned so that it can outlive the stream.
std::shared_ptr<InputStream> OpenTestFile(std::unique_ptr<UniqueTestDirectory>* dir,
                                          const std::string& filename, const std::string& content) {
    *dir = UniqueTestDirectory::Create();
    EXPECT_TRUE(*dir);
    std::string path = (*dir)->Str() + "/" + filename;
    std::ofstream file(path, std::ios::binary);
    EXPECT_TRUE(file.is_open());
    file.write(content.data(), content.size());
    EXPECT_FALSE(file.fail());
    file.close();

    EXPECT_OK_AND_ASSIGN(std::unique_ptr<FileSystem> fs, FileSystemFactory::Get("local", path, {}));
    EXPECT_OK_AND_ASSIGN(std::unique_ptr<InputStream> in, fs->Open(path));
    return std::move(in);
}

// Create a test file with the given content and return a ready ReadAheadCache
// on it. `file_size` defaults to 0, i.e. unknown, which keeps the block cache
// off: a read that no registered range covers stays a plain miss. The block
// cache tests pass the real size of the file.
std::shared_ptr<ReadAheadCache> CreateTestFileAndCache(const std::string& filename,
                                                       const std::string& content,
                                                       const CacheConfig& config,
                                                       std::vector<ByteRange> ranges,
                                                       uint64_t file_size = 0) {
    std::unique_ptr<UniqueTestDirectory> dir;
    std::shared_ptr<InputStream> in = OpenTestFile(&dir, filename, content);
    std::shared_ptr<ReadAheadCache> cache =
        std::make_shared<ReadAheadCache>(in, config, file_size, TestPool());
    EXPECT_OK(cache->AddRanges(std::move(ranges)).status());
    return cache;
}

// Assert that reading the range is a cache hit filling the destination with
// the expected content.
void AssertReadEquals(const ByteRange& range, const std::string& expected, ReadAheadCache* cache) {
    std::string dest(std::max<size_t>(range.length, 1), 'X');
    bool hit = false;
    ASSERT_OK_AND_ASSIGN(hit, cache->Read(range, dest.data()));
    ASSERT_TRUE(hit) << expected;
    ASSERT_EQ(expected, std::string_view(dest.data(), range.length));
}

// Assert that reading the range misses and leaves the destination untouched.
void AssertReadMiss(const ByteRange& range, ReadAheadCache* cache) {
    std::string dest(std::max<size_t>(range.length, 1), 'X');
    bool hit = true;
    ASSERT_OK_AND_ASSIGN(hit, cache->Read(range, dest.data()));
    ASSERT_FALSE(hit);
    ASSERT_EQ(std::string(dest.size(), 'X'), dest);
}

// Report the counter of the given metric name, or zero when it was not collected.
uint64_t CounterOf(const char* name, ReadAheadCache* cache) {
    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache->CollectMetrics(&metrics);
    return metrics->GetCounter(name).value_or(0);
}

TEST(TestReadAheadCache, TestBasics) {
    CacheConfig config =
        TestCacheConfig(/*range_size_limit=*/10,
                        /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024 * 1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr = CreateTestFileAndCache(
        "data_file", content, config,
        {{1, 2}, {3, 2}, {8, 2}, {10, 4}, {14, 0}, {15, 4}, {20, 2}, {25, 0}});
    ReadAheadCache& cache = *cache_ptr;

    AssertReadEquals({20, 2}, "uv", &cache);
    AssertReadEquals({1, 2}, "bc", &cache);
    AssertReadEquals({3, 2}, "de", &cache);
    AssertReadEquals({8, 2}, "ij", &cache);
    AssertReadEquals({10, 4}, "klmn", &cache);
    AssertReadEquals({15, 4}, "pqrs", &cache);
    AssertReadEquals({19, 3}, "tuv", &cache);

    // Zero-sized reads are immediate hits touching nothing.
    AssertReadEquals({14, 0}, "", &cache);
    AssertReadEquals({25, 0}, "", &cache);

    // Non-cached ranges miss and leave the destination untouched.
    AssertReadMiss({20, 3}, &cache);
    AssertReadMiss({0, 3}, &cache);
    AssertReadMiss({25, 2}, &cache);
}

// Test that a read spanning several adjacent cache entries is served from the
// contiguous run of entries and counted as a single hit.
TEST(TestReadAheadCache, TestMultiSegmentContiguousHit) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    // A single 25-byte range exceeds range_size_limit, so AddRanges() coalesces and splits it
    // into three adjacent entries: {0,10}, {10,10} and {20,5}.
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 25}});
    ReadAheadCache& cache = *cache_ptr;

    // Spans all three entries.
    AssertReadEquals({5, 20}, "fghijklmnopqrstuvwxy", &cache);

    // Spans the first two entries only, trimming both ends of the run.
    AssertReadEquals({5, 10}, "fghijklmno", &cache);

    // Runs past the end of the last entry: no contiguous cover, a miss.
    AssertReadMiss({5, 21}, &cache);

    // A multi-segment hit counts once with the full requested length.
    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache.CollectMetrics(&metrics);
    ASSERT_OK_AND_ASSIGN(uint64_t hits, metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
    ASSERT_EQ(hits, 2u);
    ASSERT_OK_AND_ASSIGN(uint64_t hit_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES));
    ASSERT_EQ(hit_bytes, 30u);
    ASSERT_OK_AND_ASSIGN(uint64_t misses, metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
    ASSERT_EQ(misses, 1u);
    ASSERT_OK_AND_ASSIGN(uint64_t miss_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_MISS_BYTES));
    ASSERT_EQ(miss_bytes, 21u);
}

// Test repeated reads to the same range to ensure cache reuse.
TEST(TestReadAheadCache, TestRepeatedReadCacheReuse) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/64);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}, {7, 5}});
    ReadAheadCache& cache = *cache_ptr;

    AssertReadEquals({0, 5}, "abcde", &cache);
    AssertReadEquals({0, 5}, "abcde", &cache);
}

// The cache never evicts: every prefetched range stays cached until
// ReleaseBuffers()/Reset(), regardless of how much data accumulates.
TEST(TestReadAheadCache, TestNoEvictionKeepsAllRanges) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/5, /*hole_size_limit=*/2,
                                         /*pre_buffer_limit=*/10);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}, {16, 5}});
    ReadAheadCache& cache = *cache_ptr;

    AssertReadEquals({0, 5}, "abcde", &cache);

    // Reading further ranges keeps the earlier ones cached.
    AssertReadEquals({8, 5}, "ijklm", &cache);
    AssertReadEquals({16, 5}, "qrstu", &cache);
    AssertReadEquals({0, 5}, "abcde", &cache);
}

// Test that Read() hits and misses are recorded in the cache metrics.
TEST(TestReadAheadCache, TestMetrics) {
    CacheConfig config =
        TestCacheConfig(/*range_size_limit=*/10,
                        /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024 * 1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}});
    ReadAheadCache& cache = *cache_ptr;

    AssertReadEquals({0, 5}, "abcde", &cache);
    // Out of any cached range: a miss.
    AssertReadMiss({20, 3}, &cache);

    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache.CollectMetrics(&metrics);
    // Both Read() requests are counted, regardless of hit or miss.
    ASSERT_OK_AND_ASSIGN(uint64_t read_count,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
    ASSERT_EQ(read_count, 2u);
    ASSERT_OK_AND_ASSIGN(uint64_t read_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_BYTES));
    ASSERT_EQ(read_bytes, 8u);
    ASSERT_OK_AND_ASSIGN(uint64_t hits, metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
    ASSERT_EQ(hits, 1u);
    ASSERT_OK_AND_ASSIGN(uint64_t hit_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES));
    ASSERT_EQ(hit_bytes, 5u);
    ASSERT_OK_AND_ASSIGN(uint64_t misses, metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
    ASSERT_EQ(misses, 1u);
    ASSERT_OK_AND_ASSIGN(uint64_t miss_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_MISS_BYTES));
    ASSERT_EQ(miss_bytes, 3u);
    // The hit prefetches both pending ranges in one window: two IO requests
    // for 10 bytes in total; the miss issues no further fetch.
    ASSERT_OK_AND_ASSIGN(uint64_t io_count, metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
    ASSERT_EQ(io_count, 2u);
    ASSERT_OK_AND_ASSIGN(uint64_t io_bytes, metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
    ASSERT_EQ(io_bytes, 10u);
}

// Test that ReleaseBuffers() drops the cached data but keeps the hit/miss counters
// readable, while Reset() zeroes them as well.
TEST(TestReadAheadCache, TestReleaseBuffersKeepsMetrics) {
    CacheConfig config =
        TestCacheConfig(/*range_size_limit=*/10,
                        /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024 * 1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}});
    ReadAheadCache& cache = *cache_ptr;

    AssertReadEquals({0, 5}, "abcde", &cache);

    cache.ReleaseBuffers();

    // The previously cached range is gone: the read now misses.
    AssertReadMiss({0, 5}, &cache);

    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache.CollectMetrics(&metrics);
    // The read counters survive ReleaseBuffers() as well.
    ASSERT_OK_AND_ASSIGN(uint64_t read_count,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
    ASSERT_EQ(read_count, 2u);
    ASSERT_OK_AND_ASSIGN(uint64_t read_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_BYTES));
    ASSERT_EQ(read_bytes, 10u);
    ASSERT_OK_AND_ASSIGN(uint64_t hits, metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
    ASSERT_EQ(hits, 1u);
    ASSERT_OK_AND_ASSIGN(uint64_t hit_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES));
    ASSERT_EQ(hit_bytes, 5u);
    ASSERT_OK_AND_ASSIGN(uint64_t misses, metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
    ASSERT_EQ(misses, 1u);
    // The io counters survive ReleaseBuffers() as well.
    ASSERT_OK_AND_ASSIGN(uint64_t io_count, metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
    ASSERT_EQ(io_count, 1u);
    ASSERT_OK_AND_ASSIGN(uint64_t io_bytes, metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
    ASSERT_EQ(io_bytes, 5u);

    // Reset() clears the counters too.
    cache.Reset();
    std::shared_ptr<Metrics> reset_metrics = std::make_shared<MetricsImpl>();
    cache.CollectMetrics(&reset_metrics);
    ASSERT_OK_AND_ASSIGN(read_count, reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
    ASSERT_EQ(read_count, 0u);
    ASSERT_OK_AND_ASSIGN(hits, reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
    ASSERT_EQ(hits, 0u);
    ASSERT_OK_AND_ASSIGN(misses, reset_metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
    ASSERT_EQ(misses, 0u);
    ASSERT_OK_AND_ASSIGN(io_count, reset_metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
    ASSERT_EQ(io_count, 0u);
    ASSERT_OK_AND_ASSIGN(io_bytes, reset_metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
    ASSERT_EQ(io_bytes, 0u);
}

// Test that a failed prefetch surfaces as an error Status from Read(), not as
// a miss: the entry exists from the moment its fetch is submitted and its
// future carries the IO error.
TEST(TestReadAheadCache, TestPrefetchIOErrorPropagation) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    auto io_hook = paimon::IOHook::GetInstance();

    // Single entry: the prefetch is the first IO after the hook is armed.
    {
        std::shared_ptr<ReadAheadCache> cache_ptr =
            CreateTestFileAndCache("data_file", content, config, {{0, 10}});
        paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
        io_hook->Reset(0, paimon::IOHook::Mode::RETURN_ERROR);
        std::string dest(5, 'X');
        ASSERT_NOK_WITH_MSG(cache_ptr->Read({0, 5}, dest.data()),
                            "io hook triggered io error at position");
    }

    // Several adjacent entries: the error of any segment aborts the read.
    {
        std::shared_ptr<ReadAheadCache> cache_ptr =
            CreateTestFileAndCache("data_file", content, config, {{0, 25}});
        paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
        io_hook->Reset(1, paimon::IOHook::Mode::RETURN_ERROR);
        std::string dest(20, 'X');
        ASSERT_NOK_WITH_MSG(cache_ptr->Read({0, 20}, dest.data()),
                            "io hook triggered io error at position");
    }
}

// Test that Warmup() fetches the pending ranges up front so the first Read()
// issues no further IO, while without Warmup() the first Read() triggers the
// prefetch itself.
TEST(TestReadAheadCache, TestWarmupPrefetchesBeforeFirstRead) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache1 =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}});
    cache1->Warmup();
    std::shared_ptr<ReadAheadCache> cache2 =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}});

    auto io_hook = paimon::IOHook::GetInstance();
    paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
    // Any new IO fails: the warmed-up reads must be served without fetching.
    io_hook->Reset(0, paimon::IOHook::Mode::RETURN_ERROR);

    AssertReadEquals({0, 5}, "abcde", cache1.get());
    AssertReadEquals({8, 5}, "ijklm", cache1.get());

    // Without Warmup() the first Read() starts the prefetch and sees the error.
    std::string dest(5, 'X');
    ASSERT_NOK(cache2->Read({0, 5}, dest.data()));
}

// Warmup() without any pending ranges is a safe no-op.
TEST(TestReadAheadCache, TestWarmupWithEmptyRanges) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {});
    cache_ptr->Warmup();
    AssertReadMiss({0, 5}, cache_ptr.get());
}

// A reader racing an in-flight prefetch must find the published entry and wait
// on its future instead of missing and re-fetching the same bytes: entries are
// published under the lock before their fetch is dispatched.
TEST(TestReadAheadCache, TestInFlightEntryServesRacingReader) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::unique_ptr<UniqueTestDirectory> dir;
    auto gated = std::make_shared<GatedAsyncInputStream>(OpenTestFile(&dir, "data_file", content));

    ReadAheadCache cache(gated, config, /*file_size=*/0, TestPool());
    ASSERT_OK(cache.AddRanges({{0, 5}}).status());
    cache.Warmup();

    // The prefetch entry is published, but its fetch is still held.
    ASSERT_EQ(gated->AsyncReadCount(), 1);

    // A racing reader blocks on the in-flight entry's future and is served
    // from it once the fetch completes, without triggering a second fetch.
    std::thread reader([&cache, &gated]() {
        std::string dest(5, 'X');
        Result<bool> res = cache.Read({0, 5}, dest.data());
        EXPECT_TRUE(res.ok());
        if (res.ok()) {
            EXPECT_TRUE(res.value());
        }
        EXPECT_EQ("abcde", std::string_view(dest.data(), 5));
        EXPECT_EQ(gated->AsyncReadCount(), 1);
    });
    // Give the reader time to block on the in-flight entry's future before
    // completing the fetch.
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    gated->ReleaseAll();
    reader.join();

    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache.CollectMetrics(&metrics);
    ASSERT_OK_AND_ASSIGN(uint64_t misses, metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
    ASSERT_EQ(misses, 0u);
    ASSERT_OK_AND_ASSIGN(uint64_t io_count, metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
    ASSERT_EQ(io_count, 1u);
}

// An object store stream destroys the callback of a read after it has resolved
// it, so the last reference to a prefetch buffer can be dropped by an IO thread
// after the cache - and the memory pool it holds - is already gone. The buffer
// must keep its pool alive until then, or it frees its allocation against a
// destroyed pool.
TEST(TestReadAheadCache, TestPrefetchBufferKeepsPoolAliveAfterCacheIsGone) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::unique_ptr<UniqueTestDirectory> dir;
    auto gated = std::make_shared<GatedAsyncInputStream>(
        OpenTestFile(&dir, "data_file", "abcdefghijklmnopqrstuvwxyz"));
    std::weak_ptr<MemoryPool> weak_pool;
    {
        std::shared_ptr<MemoryPool> pool = TestPool();
        weak_pool = pool;
        // Declared after the pool, so the cache is destroyed before the last
        // reference of this test to the pool is dropped.
        ReadAheadCache cache(gated, config, /*file_size=*/0, pool);
        ASSERT_OK(cache.AddRanges({{0, 5}}).status());
        cache.Warmup();
        ASSERT_EQ(gated->AsyncReadCount(), 1);
        // The fetch is completed, but the stream keeps its callback - and with it
        // a reference to the entry buffer - alive.
        gated->ReleaseAllKeepingCallbacks();
    }

    // The cache and the pool reference of this test are gone, so the callback
    // holds the last reference to the buffer, which must still hold the pool.
    ASSERT_FALSE(weak_pool.expired());
    gated->DropCompletedCallbacks();
    ASSERT_TRUE(weak_pool.expired());
}

// Test that pre_buffer_limit truncates the prefetch window: only ranges within
// the window are fetched at once, later reads fetch the remaining batches.
TEST(TestReadAheadCache, TestPreBufferWindowLimit) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/10);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 10}, {16, 10}});
    ReadAheadCache& cache = *cache_ptr;

    auto io_hook = paimon::IOHook::GetInstance();
    paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
    // IOCount() only counts while armed; INT64_MAX never triggers the error mode.
    io_hook->Reset(INT64_MAX, paimon::IOHook::Mode::RETURN_ERROR);

    AssertReadEquals({0, 10}, "abcdefghij", &cache);
    // The second range did not fit into the window: only one prefetch IO.
    ASSERT_EQ(io_hook->IOCount(), 1);

    AssertReadEquals({16, 10}, "qrstuvwxyz", &cache);
    // The second read triggered exactly one more prefetch IO.
    ASSERT_EQ(io_hook->IOCount(), 2);

    // The range is cached now: re-reading it issues no IO at all.
    io_hook->Reset(INT64_MAX, paimon::IOHook::Mode::RETURN_ERROR);
    AssertReadEquals({16, 10}, "qrstuvwxyz", &cache);
    ASSERT_EQ(io_hook->IOCount(), 0);
}

// Test that the cache can be re-initialized after Reset() and serves the new ranges.
TEST(TestReadAheadCache, TestReinitAfterReset) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}});
    ReadAheadCache& cache = *cache_ptr;

    AssertReadEquals({0, 5}, "abcde", &cache);

    cache.Reset();
    ASSERT_OK(cache.AddRanges({{3, 4}}).status());
    AssertReadEquals({3, 4}, "defg", &cache);

    // The old ranges are gone.
    AssertReadMiss({20, 2}, &cache);
}

TEST(TestReadAheadCache, TestAddRangesRejectsOutOfFileRanges) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/1024,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/2048);
    auto cache = CreateTestFileAndCache("data_file", std::string(1024, 'x'), config, {},
                                        /*file_size=*/1024);
    constexpr uint64_t kMaxValue = std::numeric_limits<uint64_t>::max();
    for (const ByteRange& range : std::vector<ByteRange>{{512, 32 * 1024 * 1024},
                                                         {1024, 1},
                                                         {1025, 0},
                                                         {1023, 2},
                                                         {kMaxValue, 1},
                                                         {1, kMaxValue}}) {
        ASSERT_NOK_WITH_MSG(cache->AddRanges({range}), "exceeds file size 1024");
    }
}

TEST(TestReadAheadCache, TestInvalidLateRangesPreserveRegisteredRanges) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/8,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/32);
    config.SetBlockCacheLimit(0);
    auto cache = CreateTestFileAndCache("data_file", std::string(32, 'x'), config, {{0, 8}},
                                        /*file_size=*/32);
    AssertReadEquals({0, 8}, std::string(8, 'x'), cache.get());

    ASSERT_NOK_WITH_MSG(cache->AddRanges({{16, 8}, {31, 2}}), "exceeds file size 32");

    AssertReadEquals({0, 8}, std::string(8, 'x'), cache.get());
    AssertReadMiss({16, 8}, cache.get());
}

TEST(TestReadAheadCache, TestAddRangesAcceptsFileEndBoundary) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/4,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/16);
    auto cache = CreateTestFileAndCache("data_file", "abcdefgh", config, {{4, 4}},
                                        /*file_size=*/8);
    AssertReadEquals({4, 4}, "efgh", cache.get());
    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added, cache->AddRanges({{8, 0}}));
    ASSERT_FALSE(first_added.has_value());
    ASSERT_OK(cache->AddRanges({{0, 8}}).status());
    AssertReadEquals({0, 8}, "abcdefgh", cache.get());
}

TEST(TestReadAheadCache, TestAddRangesPreservesUnknownFileSize) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/1024 * 1024,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/2048);
    auto cache = CreateTestFileAndCache("data_file", std::string(1024, 'x'), config, {},
                                        /*file_size=*/0);
    // Zero retains its existing unknown-size meaning. Registration performs no I/O.
    ASSERT_OK(cache->AddRanges({{512, 32 * 1024 * 1024}}).status());
}

// Test that AddRanges() registers ranges that only become known after the cache was
// initialized, so the reads that follow are served from the cache.
TEST(TestReadAheadCache, TestAddRangesRegistersNewRanges) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 10}});
    ReadAheadCache& cache = *cache_ptr;

    AssertReadMiss({16, 10}, &cache);

    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added, cache.AddRanges({{16, 10}}));
    ASSERT_TRUE(first_added.has_value());
    ASSERT_EQ(16u, first_added.value());

    AssertReadEquals({16, 10}, "qrstuvwxyz", &cache);
    // The range registered first is still served.
    AssertReadEquals({0, 10}, "abcdefghij", &cache);
}

// Of a new range intersecting an already registered one, only the part the registered ranges do
// not cover is registered: the registered ranges stay disjoint, and the read the new range asked
// for is served instead of dropped along with it.
TEST(TestReadAheadCache, TestAddRangesClipsIntersectingRange) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 10}});
    ReadAheadCache& cache = *cache_ptr;

    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added, cache.AddRanges({{5, 10}}));
    ASSERT_TRUE(first_added.has_value());
    // [5, 10) is already registered, so the registration starts at the first byte that is not.
    ASSERT_EQ(10u, first_added.value());

    // Served by the registered range and the newly registered one together, as they are adjacent.
    AssertReadEquals({5, 10}, "fghijklmno", &cache);
    AssertReadEquals({0, 10}, "abcdefghij", &cache);
}

// A new range fully covered by the registered ones registers nothing at all.
TEST(TestReadAheadCache, TestAddRangesDropsCoveredRange) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 10}});
    ReadAheadCache& cache = *cache_ptr;

    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added, cache.AddRanges({{2, 4}}));
    ASSERT_FALSE(first_added.has_value());

    AssertReadEquals({0, 10}, "abcdefghij", &cache);
}

// The registered ranges stay offset-ordered and disjoint when the new ranges interleave with them,
// so that a read still finds its covering entries.
TEST(TestReadAheadCache, TestAddRangesInterleavesWithRegisteredRanges) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/26,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{4, 4}, {16, 4}});
    ReadAheadCache& cache = *cache_ptr;

    // Two ranges each straddling a registered one, plus one entirely in the hole between them.
    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added,
                         cache.AddRanges({{2, 4}, {10, 2}, {18, 6}}));
    ASSERT_TRUE(first_added.has_value());
    ASSERT_EQ(2u, first_added.value());

    AssertReadEquals({2, 6}, "cdefgh", &cache);
    AssertReadEquals({10, 2}, "kl", &cache);
    AssertReadEquals({16, 8}, "qrstuvwx", &cache);
    // The holes nobody registered are still misses.
    AssertReadMiss({12, 4}, &cache);
}

// A registered range larger than the range size limit is cut into pieces, so it is fetched by
// several concurrent requests rather than by one long one. The pieces are adjacent, so a read
// spanning them is still served.
TEST(TestReadAheadCache, TestAddRangesSplitsLargeRange) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/5,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}});
    ReadAheadCache& cache = *cache_ptr;

    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added, cache.AddRanges({{10, 15}}));
    ASSERT_TRUE(first_added.has_value());
    ASSERT_EQ(10u, first_added.value());

    cache.Warmup(first_added.value());
    // One request per piece, instead of one request for the whole 15 bytes.
    ASSERT_EQ(3u, CounterOf(ReadAheadCacheMetrics::IO_COUNT, &cache));

    AssertReadEquals({10, 15}, "klmnopqrstuvwxy", &cache);
    ASSERT_EQ(3u, CounterOf(ReadAheadCacheMetrics::IO_COUNT, &cache));
}

// AddRanges() rebuilds the per-range cached flags, so a range fetched before the call is not
// fetched a second time.
TEST(TestReadAheadCache, TestAddRangesKeepsCachedRanges) {
    // A window of exactly one range, so that reading the first one never reaches the second.
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/10);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 10}});
    ReadAheadCache& cache = *cache_ptr;

    cache.Warmup();
    AssertReadEquals({0, 10}, "abcdefghij", &cache);

    auto io_count_of = [&cache]() { return CounterOf(ReadAheadCacheMetrics::IO_COUNT, &cache); };
    const uint64_t io_count_before = io_count_of();
    ASSERT_EQ(1u, io_count_before);

    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added, cache.AddRanges({{16, 10}}));
    ASSERT_TRUE(first_added.has_value());

    AssertReadEquals({0, 10}, "abcdefghij", &cache);
    ASSERT_EQ(io_count_before, io_count_of());
}

// Re-registering a range the cache already covers adds nothing, and must leave the published entry
// intact: the earlier fetch stays valid, so a later read of the same range is still a hit and
// issues no further IO. A no-op registration that dropped the buffer would turn this read into a
// miss (and lose the future to wait on an in-flight fetch).
TEST(TestReadAheadCache, TestReRegisteringCoveredRangeKeepsCache) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/10);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 10}});
    ReadAheadCache& cache = *cache_ptr;

    cache.Warmup();
    AssertReadEquals({0, 10}, "abcdefghij", &cache);
    auto io_count_of = [&cache]() { return CounterOf(ReadAheadCacheMetrics::IO_COUNT, &cache); };
    const uint64_t io_count_before = io_count_of();
    ASSERT_EQ(1u, io_count_before);

    // [2,6) is already covered by [0,10), so nothing new is registered.
    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added, cache.AddRanges({{2, 6}}));
    ASSERT_FALSE(first_added.has_value());

    // The published entry survived the no-op registration: still a hit, no new IO.
    AssertReadEquals({0, 10}, "abcdefghij", &cache);
    ASSERT_EQ(io_count_before, io_count_of());
}

// An empty registration is likewise a no-op that keeps the already published entries.
TEST(TestReadAheadCache, TestEmptyAddRangesKeepsCache) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/10);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 10}});
    ReadAheadCache& cache = *cache_ptr;

    cache.Warmup();
    AssertReadEquals({0, 10}, "abcdefghij", &cache);
    auto io_count_of = [&cache]() { return CounterOf(ReadAheadCacheMetrics::IO_COUNT, &cache); };
    const uint64_t io_count_before = io_count_of();
    ASSERT_EQ(1u, io_count_before);

    ASSERT_OK_AND_ASSIGN(std::optional<uint64_t> first_added, cache.AddRanges({}));
    ASSERT_FALSE(first_added.has_value());

    AssertReadEquals({0, 10}, "abcdefghij", &cache);
    ASSERT_EQ(io_count_before, io_count_of());
}

// Warmup(offset) starts fetching from the given range rather than from the first one, which is
// what a pass registering its ranges mid-read needs: the earlier ranges belong to the pass that
// has already run.
TEST(TestReadAheadCache, TestWarmupFromOffset) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}, {16, 5}});
    ReadAheadCache& cache = *cache_ptr;

    auto io_hook = paimon::IOHook::GetInstance();
    paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
    // IOCount() only counts while armed; INT64_MAX never triggers the error mode.
    io_hook->Reset(INT64_MAX, paimon::IOHook::Mode::RETURN_ERROR);

    cache.Warmup(/*from_offset=*/16);
    // Only the range at 16 was fetched: the two before it were skipped.
    ASSERT_EQ(io_hook->IOCount(), 1);

    AssertReadEquals({16, 5}, "qrstu", &cache);
    ASSERT_EQ(io_hook->IOCount(), 1);
}

// Test that AddRanges() merges ranges separated by a small hole, so a read
// spanning the hole is served by the single coalesced entry.
TEST(TestReadAheadCache, TestCoalescesSmallHoles) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/1024,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    // Byte 5 sits in a 1-byte hole, within hole_size_limit: one entry {0,11}.
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}, {6, 5}});
    ReadAheadCache& cache = *cache_ptr;

    AssertReadEquals({4, 3}, "efg", &cache);
}

// CollectMetrics() with a null metrics output is a safe no-op.
TEST(TestReadAheadCache, TestCollectMetricsWithNullMetrics) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10,
                                         /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024);
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {{0, 5}});
    cache_ptr->CollectMetrics(/*metrics=*/nullptr);
    std::shared_ptr<Metrics> null_metrics;
    cache_ptr->CollectMetrics(&null_metrics);
}

// Read the io counters of the cache.
void GetIOCounters(ReadAheadCache* cache, uint64_t* io_count, uint64_t* io_bytes) {
    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache->CollectMetrics(&metrics);
    ASSERT_OK_AND_ASSIGN(*io_count, metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
    ASSERT_OK_AND_ASSIGN(*io_bytes, metrics->GetCounter(ReadAheadCacheMetrics::IO_BYTES));
}

// Counters of the block cache, the companion of GetIOCounters().
struct BlockCounters {
    uint64_t hits = 0;
    uint64_t hit_bytes = 0;
    uint64_t fetches = 0;
    uint64_t fetch_bytes = 0;
};

void GetBlockCounters(ReadAheadCache* cache, BlockCounters* counters) {
    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache->CollectMetrics(&metrics);
    ASSERT_OK_AND_ASSIGN(counters->hits, metrics->GetCounter(ReadAheadCacheMetrics::BLOCK_HITS));
    ASSERT_OK_AND_ASSIGN(counters->hit_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::BLOCK_HIT_BYTES));
    ASSERT_OK_AND_ASSIGN(counters->fetches,
                         metrics->GetCounter(ReadAheadCacheMetrics::BLOCK_FETCHES));
    ASSERT_OK_AND_ASSIGN(counters->fetch_bytes,
                         metrics->GetCounter(ReadAheadCacheMetrics::BLOCK_FETCH_BYTES));
}

// A cache configuration with the block cache enabled at the given granularity.
// The registered ranges are irrelevant to the block cache tests: they read the
// bytes the parquet reader reads before any range is registered.
CacheConfig BlockCacheConfig(uint64_t block_size, uint64_t block_cache_limit) {
    CacheConfig config = TestCacheConfig(/*range_size_limit=*/10, /*hole_size_limit=*/2,
                                         /*pre_buffer_limit=*/1024);
    config.SetBlockSize(block_size);
    config.SetBlockCacheLimit(block_cache_limit);
    return config;
}

// The block cache serves the reads that no registered range covers, and such a
// read is counted as a block hit rather than as a miss. See FileBlockCache and
// its own test for the block semantics themselves.
TEST(TestReadAheadCache, TestBlockCacheServesUncoveredReads) {
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    // Blocks are aligned to the end of the file: block 0 is [18, 26).
    CacheConfig config = BlockCacheConfig(/*block_size=*/8, /*block_cache_limit=*/1024);
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {}, content.size());
    ReadAheadCache& cache = *cache_ptr;

    // The whole last block, as the footer read of a parquet file does, then two
    // small reads inside it, as the page index reads of the other readers do.
    AssertReadEquals({18, 8}, "stuvwxyz", &cache);
    AssertReadEquals({20, 2}, "uv", &cache);
    AssertReadEquals({25, 1}, "z", &cache);

    BlockCounters blocks;
    GetBlockCounters(&cache, &blocks);
    ASSERT_EQ(blocks.fetches, 1u);
    ASSERT_EQ(blocks.fetch_bytes, 8u);
    ASSERT_EQ(blocks.hits, 3u);
    ASSERT_EQ(blocks.hit_bytes, 8u + 2u + 1u);

    // The block fetches are issued to the underlying stream too, so they are
    // part of the io counters.
    uint64_t io_count = 0;
    uint64_t io_bytes = 0;
    GetIOCounters(&cache, &io_count, &io_bytes);
    ASSERT_EQ(io_count, 1u);
    ASSERT_EQ(io_bytes, 8u);

    // A block hit is neither a hit of a registered range nor a miss.
    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache.CollectMetrics(&metrics);
    ASSERT_OK_AND_ASSIGN(uint64_t read_count,
                         metrics->GetCounter(ReadAheadCacheMetrics::READ_COUNT));
    ASSERT_EQ(read_count, 3u);
    ASSERT_OK_AND_ASSIGN(uint64_t hits, metrics->GetCounter(ReadAheadCacheMetrics::READ_HITS));
    ASSERT_EQ(hits, 0u);
    ASSERT_OK_AND_ASSIGN(uint64_t misses, metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
    ASSERT_EQ(misses, 0u);
}

// A read the block cache declines stays a plain miss left to the caller.
TEST(TestReadAheadCache, TestBlockCacheDeclinedReadIsAMiss) {
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    CacheConfig config = BlockCacheConfig(/*block_size=*/8, /*block_cache_limit=*/1024);
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {}, content.size());
    ReadAheadCache& cache = *cache_ptr;

    // Larger than one block, so the block cache leaves it to the caller.
    AssertReadMiss({0, 12}, &cache);

    BlockCounters blocks;
    GetBlockCounters(&cache, &blocks);
    ASSERT_EQ(blocks.fetches, 0u);
    ASSERT_EQ(blocks.hits, 0u);
    std::shared_ptr<Metrics> metrics = std::make_shared<MetricsImpl>();
    cache.CollectMetrics(&metrics);
    ASSERT_OK_AND_ASSIGN(uint64_t misses, metrics->GetCounter(ReadAheadCacheMetrics::READ_MISSES));
    ASSERT_EQ(misses, 1u);
}

// The blocks belong to the file rather than to the registered ranges: they
// survive Reset() and are only released by ReleaseBuffers().
TEST(TestReadAheadCache, TestBlockCacheSurvivesReset) {
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    CacheConfig config = BlockCacheConfig(/*block_size=*/8, /*block_cache_limit=*/1024);
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, config, {}, content.size());
    ReadAheadCache& cache = *cache_ptr;

    AssertReadEquals({18, 4}, "stuv", &cache);

    // Reset() drops the registered ranges and zeroes the counters.
    cache.Reset();
    AssertReadEquals({18, 4}, "stuv", &cache);
    BlockCounters blocks;
    GetBlockCounters(&cache, &blocks);
    ASSERT_EQ(blocks.hits, 1u);
    ASSERT_EQ(blocks.fetches, 0u);
    uint64_t io_count = 0;
    uint64_t io_bytes = 0;
    GetIOCounters(&cache, &io_count, &io_bytes);
    ASSERT_EQ(io_count, 0u);

    // ReleaseBuffers() drops the blocks, so the same read fetches again.
    cache.ReleaseBuffers();
    AssertReadEquals({18, 4}, "stuv", &cache);
    GetBlockCounters(&cache, &blocks);
    ASSERT_EQ(blocks.hits, 2u);
    ASSERT_EQ(blocks.fetches, 1u);
}

// A zero block cache limit or an unknown file size leaves the block cache out
// entirely: an uncovered read is a plain miss left to the caller.
TEST(TestReadAheadCache, TestBlockCacheDisabled) {
    std::string content = "abcdefghijklmnopqrstuvwxyz";
    CacheConfig zero_limit = BlockCacheConfig(/*block_size=*/8, /*block_cache_limit=*/0);
    std::shared_ptr<ReadAheadCache> cache_ptr =
        CreateTestFileAndCache("data_file", content, zero_limit, {}, content.size());
    AssertReadMiss({18, 4}, cache_ptr.get());
    AssertReadMiss({18, 4}, cache_ptr.get());
    BlockCounters blocks;
    GetBlockCounters(cache_ptr.get(), &blocks);
    ASSERT_EQ(blocks.fetches, 0u);
    ASSERT_EQ(blocks.hits, 0u);

    // An unknown file size cannot be aligned to, so it disables the cache too.
    CacheConfig enabled = BlockCacheConfig(/*block_size=*/8, /*block_cache_limit=*/1024);
    std::shared_ptr<ReadAheadCache> unknown_size_cache =
        CreateTestFileAndCache("data_file", content, enabled, {}, /*file_size=*/0);
    AssertReadMiss({18, 4}, unknown_size_cache.get());
    GetBlockCounters(unknown_size_cache.get(), &blocks);
    ASSERT_EQ(blocks.fetches, 0u);
    ASSERT_EQ(blocks.hits, 0u);
}

}  // namespace paimon::test