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