| /* |
| * 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 <mutex> |
| #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/testharness.h" |
| |
| namespace paimon::test { |
| |
| // Helper to create a test file, write content, and return a ready ReadAheadCache. |
| struct TestCacheEnv { |
| std::string path; |
| std::shared_ptr<paimon::ReadAheadCache> cache; |
| std::shared_ptr<paimon::MemoryPool> pool; |
| }; |
| |
| TestCacheEnv CreateTestFileAndCache(const std::string& filename, const std::string& content, |
| const paimon::CacheConfig& config, |
| std::vector<paimon::ByteRange> ranges) { |
| auto 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(); |
| |
| auto fs_result = FileSystemFactory::Get("local", path, {}); |
| EXPECT_TRUE(fs_result.ok()); |
| auto fs = std::move(fs_result).value(); |
| auto in_result = fs->Open(path); |
| EXPECT_TRUE(in_result.ok()); |
| auto in = std::move(in_result).value(); |
| |
| auto pool = GetDefaultPool(); |
| auto cache = std::make_shared<ReadAheadCache>(std::move(in), config, pool); |
| EXPECT_OK(cache->Init(std::move(ranges))); |
| return {path, cache, pool}; |
| } |
| |
| // 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; |
| EXPECT_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); |
| EXPECT_EQ(std::string(dest.size(), 'X'), dest); |
| } |
| |
| // An InputStream wrapper that holds ReadAsync callbacks until ReleaseAll() is |
| // called, letting tests observe the cache while prefetch IOs are in flight. |
| class GatedAsyncInputStream : public InputStream { |
| public: |
| explicit GatedAsyncInputStream(std::shared_ptr<InputStream> inner) : inner_(std::move(inner)) {} |
| |
| Status Close() override { |
| return inner_->Close(); |
| } |
| Status Seek(int64_t offset, SeekOrigin origin) override { |
| return inner_->Seek(offset, origin); |
| } |
| Result<int64_t> GetPos() const override { |
| return inner_->GetPos(); |
| } |
| Result<int64_t> Read(char* buffer, int64_t size) override { |
| return inner_->Read(buffer, size); |
| } |
| Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override { |
| return inner_->Read(buffer, size, offset); |
| } |
| void ReadAsync(char* buffer, int64_t size, int64_t offset, |
| std::function<void(Status)>&& callback) override { |
| std::lock_guard<std::mutex> lock(mutex_); |
| async_read_count_++; |
| pending_.push_back({buffer, size, offset, std::move(callback)}); |
| } |
| Result<std::string> GetUri() const override { |
| return inner_->GetUri(); |
| } |
| Result<int64_t> Length() const override { |
| return inner_->Length(); |
| } |
| |
| int AsyncReadCount() { |
| std::lock_guard<std::mutex> lock(mutex_); |
| return async_read_count_; |
| } |
| |
| /// Complete all held fetches against the underlying stream. |
| void ReleaseAll() { |
| std::vector<PendingRead> taken; |
| { |
| std::lock_guard<std::mutex> lock(mutex_); |
| taken = std::move(pending_); |
| pending_.clear(); |
| } |
| for (auto& read : taken) { |
| Result<int64_t> res = inner_->Read(read.buffer, read.size, read.offset); |
| read.callback(res.ok() ? Status::OK() : res.status()); |
| } |
| } |
| |
| private: |
| struct PendingRead { |
| char* buffer; |
| int64_t size; |
| int64_t offset; |
| std::function<void(Status)> callback; |
| }; |
| |
| std::shared_ptr<InputStream> inner_; |
| std::mutex mutex_; |
| std::vector<PendingRead> pending_; |
| int async_read_count_ = 0; |
| }; |
| |
| TEST(TestReadAheadCache, TestBasics) { |
| CacheConfig config(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024 * 1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache( |
| "data_file", content, config, |
| {{1, 2}, {3, 2}, {8, 2}, {10, 4}, {14, 0}, {15, 4}, {20, 2}, {25, 0}}); |
| auto& cache = *env.cache; |
| |
| 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(/*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 Init() coalesces it |
| // into three adjacent entries: {0,10}, {10,10} and {20,5}. |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 25}}); |
| auto& cache = *env.cache; |
| |
| // 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(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/64); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}, {7, 5}}); |
| auto& cache = *env.cache; |
| |
| 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(/*range_size_limit=*/5, /*hole_size_limit=*/2, |
| /*pre_buffer_limit=*/10); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}, {16, 5}}); |
| auto& cache = *env.cache; |
| |
| 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(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024 * 1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}}); |
| auto& cache = *env.cache; |
| |
| 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(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024 * 1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}}); |
| auto& cache = *env.cache; |
| |
| 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(/*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. |
| { |
| auto env = 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(env.cache->Read({0, 5}, dest.data()), |
| "io hook triggered io error at position"); |
| } |
| |
| // Several adjacent entries: the error of any segment aborts the read. |
| { |
| auto env = 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(env.cache->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(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env1 = CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}}); |
| env1.cache->Warmup(); |
| auto env2 = 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", env1.cache.get()); |
| AssertReadEquals({8, 5}, "ijklm", env1.cache.get()); |
| |
| // Without Warmup() the first Read() starts the prefetch and sees the error. |
| std::string dest(5, 'X'); |
| ASSERT_NOK(env2.cache->Read({0, 5}, dest.data())); |
| } |
| |
| // Warmup() without any pending ranges is a safe no-op. |
| TEST(TestReadAheadCache, TestWarmupWithEmptyRanges) { |
| CacheConfig config(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {}); |
| env.cache->Warmup(); |
| AssertReadMiss({0, 5}, env.cache.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(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto dir = UniqueTestDirectory::Create(); |
| ASSERT_TRUE(dir); |
| std::string path = dir->Str() + "/data_file"; |
| std::ofstream file(path, std::ios::binary); |
| ASSERT_TRUE(file.is_open()); |
| file.write(content.data(), content.size()); |
| ASSERT_FALSE(file.fail()); |
| file.close(); |
| ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", path, {})); |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs->Open(path)); |
| auto gated = std::make_shared<GatedAsyncInputStream>(std::move(in)); |
| |
| ReadAheadCache cache(gated, config, GetDefaultPool()); |
| ASSERT_OK(cache.Init({{0, 5}})); |
| 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); |
| } |
| |
| // 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(/*range_size_limit=*/10, |
| /*hole_size_limit=*/0, /*pre_buffer_limit=*/10); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 10}, {16, 10}}); |
| auto& cache = *env.cache; |
| |
| 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 Init() rejects a second call until the cache is reset. |
| TEST(TestReadAheadCache, TestDoubleInit) { |
| CacheConfig config(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}}); |
| auto& cache = *env.cache; |
| |
| Status status = cache.Init({{8, 5}}); |
| ASSERT_FALSE(status.ok()); |
| |
| // The original ranges still work. |
| AssertReadEquals({0, 5}, "abcde", &cache); |
| } |
| |
| // Test that the cache can be re-initialized after Reset() and serves the new ranges. |
| TEST(TestReadAheadCache, TestReinitAfterReset) { |
| CacheConfig config(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}}); |
| auto& cache = *env.cache; |
| |
| AssertReadEquals({0, 5}, "abcde", &cache); |
| |
| cache.Reset(); |
| ASSERT_OK(cache.Init({{3, 4}})); |
| AssertReadEquals({3, 4}, "defg", &cache); |
| |
| // The old ranges are gone. |
| AssertReadMiss({20, 2}, &cache); |
| } |
| |
| // Test that Init() merges ranges separated by a small hole, so a read |
| // spanning the hole is served by the single coalesced entry. |
| TEST(TestReadAheadCache, TestInitCoalescesSmallHoles) { |
| CacheConfig config(/*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}. |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}, {6, 5}}); |
| auto& cache = *env.cache; |
| |
| AssertReadEquals({4, 3}, "efg", &cache); |
| } |
| |
| // CollectMetrics() with a null metrics output is a safe no-op. |
| TEST(TestReadAheadCache, TestCollectMetricsWithNullMetrics) { |
| CacheConfig config(/*range_size_limit=*/10, |
| /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); |
| std::string content = "abcdefghijklmnopqrstuvwxyz"; |
| auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}}); |
| env.cache->CollectMetrics(/*metrics=*/nullptr); |
| std::shared_ptr<Metrics> null_metrics; |
| env.cache->CollectMetrics(&null_metrics); |
| } |
| |
| } // namespace paimon::test |