diff --git a/include/paimon/utils/prefetch_cache_config.h b/include/paimon/utils/prefetch_cache_config.h index 4bbf1ecd3..effd65472 100644 --- a/include/paimon/utils/prefetch_cache_config.h +++ b/include/paimon/utils/prefetch_cache_config.h @@ -34,10 +34,8 @@ namespace paimon { /// ReadAheadCache to balance memory usage, I/O efficiency, and latency hiding. class PAIMON_EXPORT CacheConfig { public: - CacheConfig(); - CacheConfig(uint64_t range_size_limit, uint64_t hole_size_limit, uint64_t pre_buffer_limit); - /// Returns the maximum allowed size (in bytes) for a single cached range. + /// Defaults to 32 MiB. uint64_t GetRangeSizeLimit() const { return range_size_limit_; } @@ -47,7 +45,8 @@ class PAIMON_EXPORT CacheConfig { range_size_limit_ = range_size_limit; } - /// Returns the maximum gap size (in bytes) considered mergeable between adjacent ranges. + /// Returns the maximum gap size (in bytes) considered mergeable between + /// adjacent ranges. Defaults to 8 KiB. uint64_t GetHoleSizeLimit() const { return hole_size_limit_; } @@ -57,7 +56,8 @@ class PAIMON_EXPORT CacheConfig { hole_size_limit_ = hole_size_limit; } - /// Returns the maximum size to pre-buffer ahead of the current read position. + /// Returns the maximum size to pre-buffer ahead of the current read + /// position. Defaults to 256 MiB. uint64_t GetPreBufferLimit() const { return pre_buffer_limit_; } @@ -67,10 +67,50 @@ class PAIMON_EXPORT CacheConfig { pre_buffer_limit_ = pre_buffer_limit; } + /// Returns the granularity (in bytes) of the block cache entries serving the + /// small reads that the prefetched ranges do not cover. Defaults to 64 KiB. + uint64_t GetBlockSize() const { + return block_size_; + } + + /// Sets the granularity (in bytes) of the block cache entries. + void SetBlockSize(uint64_t block_size) { + block_size_ = block_size; + } + + /// Returns the maximum total size (in bytes) of the block cache entries of + /// one file. Zero disables the block cache. Defaults to 1 MiB. + uint64_t GetBlockCacheLimit() const { + return block_cache_limit_; + } + + /// Sets the maximum total size (in bytes) of the block cache entries of one + /// file. Zero disables the block cache. + void SetBlockCacheLimit(uint64_t block_cache_limit) { + block_cache_limit_ = block_cache_limit; + } + private: - uint64_t range_size_limit_; - uint64_t hole_size_limit_; - uint64_t pre_buffer_limit_; + // The defaults are aligned with the reader's request granularity and with + // realistic data file sizes: + // - range_size_limit matches the parquet reader's 32 MiB request blocks + // (Arrow ReadRangeCache's own range limit); a smaller limit cuts entries + // below the request size, so a request can never be served from one piece. + // - pre_buffer_limit must exceed the LARGEST single read a reader issues + // (coalesced column-chunk reads of ~128 MiB were observed): fetches are + // only dispatched up to this window, so a request reaching past it can + // never be served and falls back to a second fetch of the same bytes. + uint64_t range_size_limit_ = 32 * 1024 * 1024; + uint64_t hole_size_limit_ = 8 * 1024; + uint64_t pre_buffer_limit_ = 256 * 1024 * 1024; + // Blocks are aligned to the END of the file, so a block never reaches past + // EOF. 64 KiB matches the footer read of the parquet reader (arrow's + // kDefaultFooterReadSize), which then covers exactly one block instead of + // straddling two. + uint64_t block_size_ = 64 * 1024; + // One block is enough for the metadata tail of a file; the limit only + // bounds the pathological case, as blocks are never evicted. + uint64_t block_cache_limit_ = 1024 * 1024; }; } // namespace paimon diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 4aca72abc..aecef96d1 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -179,6 +179,7 @@ set(PAIMON_COMMON_SRCS common/data/shredding/shredding_file_reader.cpp common/utils/delta_varint_compressor.cpp common/utils/fields_comparator.cpp + common/utils/file_block_cache.cpp common/utils/path_util.cpp common/utils/range.cpp common/utils/read_ahead_cache.cpp @@ -667,6 +668,7 @@ if(PAIMON_BUILD_TESTS) common/utils/roaring_bitmap64_test.cpp common/utils/range_helper_test.cpp common/utils/read_ahead_cache_test.cpp + common/utils/file_block_cache_test.cpp common/io/cache/lru_cache_test.cpp common/io/cache/cache_manager_test.cpp common/utils/byte_range_combiner_test.cpp diff --git a/src/paimon/common/io/cache_input_stream_test.cpp b/src/paimon/common/io/cache_input_stream_test.cpp index 0b14dc33b..465d7458e 100644 --- a/src/paimon/common/io/cache_input_stream_test.cpp +++ b/src/paimon/common/io/cache_input_stream_test.cpp @@ -60,9 +60,14 @@ class CacheInputStreamTest : public ::testing::Test { std::shared_ptr CreateCache(std::vector ranges) { auto stream = OpenFile(); - CacheConfig config(/*range_size_limit=*/1024, - /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024 * 1024); - auto cache = std::make_shared(std::move(stream), config, pool_); + CacheConfig config; + config.SetRangeSizeLimit(1024); + config.SetHoleSizeLimit(0); + config.SetPreBufferLimit(1024 * 1024); + // The file size is left unknown so the block cache stays off: these + // tests exercise the fallback of CacheInputStream on a cache miss. + auto cache = + std::make_shared(std::move(stream), config, /*file_size=*/0, pool_); EXPECT_OK(cache->Init(std::move(ranges))); return cache; } @@ -204,9 +209,12 @@ TEST_F(CacheInputStreamTest, TestReadAsyncCacheReadError) { ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", file_path_, {})); ASSERT_OK_AND_ASSIGN(auto cache_stream, fs->Open(file_path_)); ASSERT_OK_AND_ASSIGN(auto underlying, fs->Open(file_path_)); - CacheConfig config(/*range_size_limit=*/1024, - /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024 * 1024); - auto cache = std::make_shared(std::move(cache_stream), config, pool_); + CacheConfig config; + config.SetRangeSizeLimit(1024); + config.SetHoleSizeLimit(0); + config.SetPreBufferLimit(1024 * 1024); + auto cache = std::make_shared(std::move(cache_stream), config, + /*file_size=*/0, pool_); ASSERT_OK(cache->Init(std::vector{{0, 10}})); // Now activate IOHook so that the prefetch IO (triggered by cache_->Read -> PreBuffer) diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp index f24dd2b5a..51589ba3c 100644 --- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp +++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp @@ -215,6 +215,9 @@ Result> PrefetchFileBatchReaderImpl if (batch_size <= 0) { return Status::Invalid("batch size should be greater than 0."); } + if (data_file_size < 0) { + return Status::Invalid("data file size should not be negative."); + } if (reader_builder == nullptr) { return Status::Invalid("reader_builder should not be nullptr."); } @@ -236,7 +239,11 @@ Result> PrefetchFileBatchReaderImpl if (io_metrics) { input_stream = std::make_shared(input_stream, io_metrics); } - cache = std::make_shared(input_stream, cache_config, pool); + // The file size lets the cache align its blocks to the end of the file, + // where the metadata the readers read before any range is registered + // lives. A zero size means unknown and disables the block cache. + cache = std::make_shared(input_stream, cache_config, + static_cast(data_file_size), pool); } std::vector>>> futures; for (uint32_t i = 0; i < prefetch_max_parallel_num; i++) { diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp index 02f9ba8d4..5e0c40d4e 100644 --- a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp +++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp @@ -664,10 +664,10 @@ TEST_F(PrefetchFileBatchReaderImplTest, WorkloopSetReadStatusWhenCacheInitFailed int32_t batch_size = 5; int32_t prefetch_max_parallel_num = 1; MockFormatReaderBuilder reader_builder(data_array, data_type_, batch_size); - CacheConfig invalid_cache_config( - /*range_size_limit=*/4 * 1024, - /*hole_size_limit=*/8 * 1024, - /*pre_buffer_limit=*/128 * 1024); + CacheConfig invalid_cache_config; + invalid_cache_config.SetRangeSizeLimit(4 * 1024); + invalid_cache_config.SetHoleSizeLimit(8 * 1024); + invalid_cache_config.SetPreBufferLimit(128 * 1024); ASSERT_OK_AND_ASSIGN( auto reader, @@ -930,6 +930,16 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) { /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/true, CacheConfig(), /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_))); } + { + ASSERT_NOK_WITH_MSG( + PrefetchFileBatchReaderImpl::Create( + data_file_path, /*data_file_size=*/-1, &reader_builder, mock_fs_, + prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2, + /*enable_adaptive_prefetch_strategy=*/false, executor_, + /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/true, CacheConfig(), + /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)), + "data file size should not be negative"); + } { ASSERT_NOK(PrefetchFileBatchReaderImpl::Create( data_file_path, /*data_file_size=*/0, &reader_builder, mock_fs_, diff --git a/src/paimon/common/utils/file_block_cache.cpp b/src/paimon/common/utils/file_block_cache.cpp new file mode 100644 index 000000000..fec0472a7 --- /dev/null +++ b/src/paimon/common/utils/file_block_cache.cpp @@ -0,0 +1,166 @@ +/* + * 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/file_block_cache.h" + +#include + +namespace paimon { + +FileBlockCache::FileBlockCache(const std::shared_ptr& stream, uint64_t file_size, + uint64_t block_size, uint64_t capacity, + const std::shared_ptr& memory_pool) + : stream_(stream), + file_size_(file_size), + block_size_(block_size), + capacity_(capacity), + memory_pool_(memory_pool) {} + +FileBlockCache::~FileBlockCache() { + // The fetches write into the block buffers, so they must not outlive the + // stream or the memory pool. + std::lock_guard lock(mutex_); + for (auto& block : blocks_) { + block.second->future.wait(); + } +} + +bool FileBlockCache::Read(const ByteRange& range, char* dest) { + if (!CanServe(range)) { + return false; + } + const uint64_t index = IndexOf(range.offset); + std::shared_ptr block; + bool dispatch = false; + { + // Publishing the promise-backed block under the lock before its fetch is + // dispatched is what makes concurrent readers of the same block wait for + // that one fetch instead of issuing their own. + std::lock_guard lock(mutex_); + auto it = blocks_.find(index); + if (it != blocks_.end()) { + block = it->second; + } else { + const ByteRange block_range = RangeOf(index); + // Blocks are never evicted, so an exhausted capacity means this read + // goes back to the caller instead of replacing a cached block. + if (cached_bytes_ + block_range.length > capacity_) { + return false; + } + block = std::make_shared(); + block->range = block_range; + block->buffer = std::make_shared(block_range.length, memory_pool_.get()); + block->promise = std::make_shared>(); + block->future = block->promise->get_future().share(); + blocks_.emplace(index, block); + cached_bytes_ += block_range.length; + dispatch = true; + } + } + if (dispatch) { + Fetch(block); + } + // Wait and copy OUTSIDE the lock, so that a reader waiting for a fetch does + // not keep the other readers out of the map. + if (!block->future.get().ok()) { + // A block fetch reads more than the caller asked for, so its failure must + // not fail the caller's read: the read goes back to the caller, which + // reports the real error itself if its own bytes cannot be read either. + // This is what keeps a block whose range the file does not have - the + // file metadata records a size larger than the physical file - from + // failing the reads of the region it covers. + // + // The block is left in place with its failed future, so the later reads + // of that block are declined without fetching it again. + return false; + } + std::memcpy(dest, block->buffer->data() + (range.offset - block->range.offset), range.length); + hits_.fetch_add(1, std::memory_order_relaxed); + hit_bytes_.fetch_add(range.length, std::memory_order_relaxed); + return true; +} + +void FileBlockCache::Release() { + std::lock_guard lock(mutex_); + // Blocks are never evicted, so waiting on blocks_ covers every dispatched + // fetch before the buffers they write into go away. + for (auto& block : blocks_) { + block.second->future.wait(); + } + blocks_.clear(); + cached_bytes_ = 0; +} + +void FileBlockCache::ResetCounters() { + hits_.store(0, std::memory_order_relaxed); + hit_bytes_.store(0, std::memory_order_relaxed); + fetches_.store(0, std::memory_order_relaxed); + fetch_bytes_.store(0, std::memory_order_relaxed); +} + +FileBlockCache::Counters FileBlockCache::GetCounters() const { + Counters counters; + counters.hits = hits_.load(std::memory_order_relaxed); + counters.hit_bytes = hit_bytes_.load(std::memory_order_relaxed); + counters.fetches = fetches_.load(std::memory_order_relaxed); + counters.fetch_bytes = fetch_bytes_.load(std::memory_order_relaxed); + return counters; +} + +bool FileBlockCache::CanServe(const ByteRange& range) const { + if (capacity_ == 0 || block_size_ == 0 || file_size_ == 0) { + return false; + } + if (range.length == 0 || range.length > block_size_) { + return false; + } + // A read reaching past EOF is left to the caller: serving it would mean + // short-reading into the block buffer. + if (range.offset >= file_size_ || range.length > file_size_ - range.offset) { + return false; + } + // A read straddling two blocks would need both of them to be present; it is + // left to the caller instead, which keeps one block per served read. + return IndexOf(range.offset) == IndexOf(range.offset + range.length - 1); +} + +uint64_t FileBlockCache::IndexOf(uint64_t offset) const { + // Counted from the end of the file, so that block 0 is the last block. + return (file_size_ - 1 - offset) / block_size_; +} + +ByteRange FileBlockCache::RangeOf(uint64_t index) const { + const uint64_t end = file_size_ - index * block_size_; + const uint64_t offset = end > block_size_ ? end - block_size_ : 0; + return {offset, end - offset}; +} + +void FileBlockCache::Fetch(const std::shared_ptr& block) { + fetches_.fetch_add(1, std::memory_order_relaxed); + fetch_bytes_.fetch_add(block->range.length, std::memory_order_relaxed); + auto promise = block->promise; + auto buffer = block->buffer; + // The buffer and the promise are captured, so the async read keeps its + // destination and the future it resolves alive. + stream_->ReadAsync(buffer->data(), static_cast(buffer->size()), + static_cast(block->range.offset), + [promise, buffer](Status status) { promise->set_value(status); }); +} + +} // namespace paimon diff --git a/src/paimon/common/utils/file_block_cache.h b/src/paimon/common/utils/file_block_cache.h new file mode 100644 index 000000000..b7971d6f3 --- /dev/null +++ b/src/paimon/common/utils/file_block_cache.h @@ -0,0 +1,145 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "paimon/common/utils/read_ahead_cache.h" +#include "paimon/fs/file_system.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/status.h" +#include "paimon/visibility.h" + +namespace paimon { + +/// A cache of fixed-size blocks of one file, serving the reads that no +/// prefetched range covers: a parquet reader reads the footer and the page index +/// before any range can be registered, and every reader of the file reads the +/// same bytes. +/// +/// Blocks are aligned to the END of the file: block 0 is +/// [file_size - block_size, file_size). The metadata of a parquet/orc file lives +/// in its tail and arrow reads exactly the last 64 KiB as the footer, so an +/// end-aligned block matches that read instead of straddling two blocks. It also +/// keeps every block inside the file, as long as the given file size is the size +/// of the file: a block fetch failing because the file is shorter than that only +/// costs the caching of that block, see Read(). +/// +/// A block is published before its fetch is dispatched, so concurrent readers of +/// the same block wait for that one fetch instead of issuing their own. +/// +/// Blocks are never evicted: once the capacity is reached Read() declines +/// instead of replacing a block. That keeps every dispatched fetch reachable +/// through its block, so Release() and the destructor can wait for the fetches +/// still writing into the block buffers. +class PAIMON_EXPORT FileBlockCache { + public: + /// Requests served by a block and fetches issued for the blocks themselves, + /// reported by the owner of this cache through its own metrics. + struct Counters { + uint64_t hits = 0; + uint64_t hit_bytes = 0; + uint64_t fetches = 0; + uint64_t fetch_bytes = 0; + }; + + /// @param stream The stream the blocks are fetched from. + /// @param file_size Size of the file behind `stream`, which the blocks are + /// aligned to the end of. Must not be zero. + /// @param block_size Granularity of the blocks. Must not be zero. + /// @param capacity Maximum total size of the cached blocks, in bytes. + /// @param memory_pool The pool the block buffers are allocated from. + FileBlockCache(const std::shared_ptr& stream, uint64_t file_size, + uint64_t block_size, uint64_t capacity, + const std::shared_ptr& memory_pool); + ~FileBlockCache(); + + /// Serve the given range out of its block, fetching that block first if it + /// is not cached yet. + /// @param range The byte range to read. + /// @param dest Destination buffer with at least `range.length` bytes. + /// @return true if the range was served and `dest` was filled; false when + /// one block cannot serve the range (it straddles two blocks, is larger than + /// a block or reaches past EOF), when the capacity is exhausted or when the + /// fetch of the block failed, leaving `dest` untouched so the caller can read + /// the bytes itself. A fetch failure is never reported to the caller: the + /// block reads more than the caller asked for, so the caller reads its own + /// bytes instead and reports the failure itself if they cannot be read + /// either. A block whose fetch failed is not fetched again. + bool Read(const ByteRange& range, char* dest); + + /// Drop all cached blocks, waiting for the fetches still writing into their + /// buffers. The counters are kept readable for the owner's metrics. + void Release(); + + /// Zero the counters while keeping the cached blocks, which cache the file + /// rather than a round of reads. + void ResetCounters(); + + Counters GetCounters() const; + + private: + /// A cached block. Blocks are handed out as shared_ptr so that a reader + /// keeps its block alive once it has released the lock. + struct Block { + ByteRange range; + std::shared_ptr buffer; + std::shared_ptr> promise; + // shared_future, as every reader of the block waits on it. + std::shared_future future; + }; + + /// Whether one block can serve the given range. + bool CanServe(const ByteRange& range) const; + /// Index of the block holding `offset`, counted from the END of the file, so + /// that block 0 is the last block. `offset` must be inside the file. + uint64_t IndexOf(uint64_t offset) const; + /// Range of the block with the given index, clamped at the start of the file. + ByteRange RangeOf(uint64_t index) const; + /// Fetch the block into its buffer and resolve its promise with the outcome. + /// Must be called after the block has been published, and only by the reader + /// that published it. + void Fetch(const std::shared_ptr& block); + + std::shared_ptr stream_; + uint64_t file_size_; + uint64_t block_size_; + uint64_t capacity_; + std::shared_ptr memory_pool_; + // Blocks are aligned, so keying them by index keeps them disjoint by + // construction and needs no ordering. A plain mutex is enough: only the + // reads that no prefetched range covers touch the map, and they are few. + mutable std::mutex mutex_; + std::unordered_map> blocks_; + // Bytes held by blocks_, guarded by mutex_ and bounded by capacity_. + uint64_t cached_bytes_ = 0; + std::atomic hits_{0}; + std::atomic hit_bytes_{0}; + std::atomic fetches_{0}; + std::atomic fetch_bytes_{0}; +}; + +} // namespace paimon diff --git a/src/paimon/common/utils/file_block_cache_test.cpp b/src/paimon/common/utils/file_block_cache_test.cpp new file mode 100644 index 000000000..f01e7ff62 --- /dev/null +++ b/src/paimon/common/utils/file_block_cache_test.cpp @@ -0,0 +1,286 @@ +/* + * 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/file_block_cache.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/common/factories/io_hook.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 { + +namespace { + +constexpr char kContent[] = "abcdefghijklmnopqrstuvwxyz"; +// 26 bytes in blocks of 8: block 0 = [18, 26), block 1 = [10, 18), +// block 2 = [2, 10) and block 3 = [0, 2), truncated at the start of the file. +constexpr uint64_t kBlockSize = 8; + +// Write the test content into a fresh directory and open it for reading. The +// directory is returned so that it outlives the stream. +std::shared_ptr OpenTestFile(std::unique_ptr* dir) { + *dir = UniqueTestDirectory::Create(); + EXPECT_TRUE(*dir); + std::string path = (*dir)->Str() + "/data_file"; + std::ofstream file(path, std::ios::binary); + EXPECT_TRUE(file.is_open()); + file.write(kContent, sizeof(kContent) - 1); + EXPECT_FALSE(file.fail()); + file.close(); + + Result> fs = FileSystemFactory::Get("local", path, {}); + EXPECT_OK(fs.status()); + Result> in = fs.value()->Open(path); + EXPECT_OK(in.status()); + return std::move(in).value(); +} + +// Assert that the range is served out of a block with the expected content. +void AssertServed(const ByteRange& range, const std::string& expected, FileBlockCache* cache) { + std::string dest(range.length, 'X'); + ASSERT_TRUE(cache->Read(range, dest.data())) << expected; + ASSERT_EQ(expected, std::string_view(dest.data(), range.length)); +} + +// Assert that the cache declines the range and leaves the destination untouched, +// so that the caller can read the bytes itself. +void AssertDeclined(const ByteRange& range, FileBlockCache* cache) { + std::string dest(range.length, 'X'); + ASSERT_FALSE(cache->Read(range, dest.data())); + ASSERT_EQ(std::string(dest.size(), 'X'), dest); +} + +} // namespace + +// The tail of the file is fetched once and then serves every later read falling +// into it, the way the footer read of a parquet file serves the page index reads +// of the readers sharing the cache. +TEST(TestFileBlockCache, TestServesRepeatedTailReads) { + std::unique_ptr dir; + FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize, + /*capacity=*/1024, GetDefaultPool()); + + // The whole last block, as the footer read of a parquet file does. + AssertServed({18, 8}, "stuvwxyz", &cache); + // Five small reads inside that block, as the page index reads do. + AssertServed({25, 1}, "z", &cache); + AssertServed({20, 2}, "uv", &cache); + AssertServed({18, 1}, "s", &cache); + AssertServed({22, 4}, "wxyz", &cache); + AssertServed({19, 3}, "tuv", &cache); + + // One fetch of one block served all six reads. + const FileBlockCache::Counters counters = cache.GetCounters(); + ASSERT_EQ(counters.fetches, 1u); + ASSERT_EQ(counters.fetch_bytes, 8u); + ASSERT_EQ(counters.hits, 6u); + ASSERT_EQ(counters.hit_bytes, 8u + 1u + 2u + 1u + 4u + 3u); +} + +// The lowest block is truncated at the start of the file, so no block fetch +// reads past either end of the file. +TEST(TestFileBlockCache, TestClampsLowestBlockAtFileStart) { + std::unique_ptr dir; + FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize, + /*capacity=*/1024, GetDefaultPool()); + + AssertServed({0, 2}, "ab", &cache); + AssertServed({1, 1}, "b", &cache); + + const FileBlockCache::Counters counters = cache.GetCounters(); + ASSERT_EQ(counters.fetches, 1u); + ASSERT_EQ(counters.fetch_bytes, 2u); + ASSERT_EQ(counters.hits, 2u); +} + +// A reader racing the fetch of a block must wait on the published block instead +// of issuing a second fetch for the same bytes. +TEST(TestFileBlockCache, TestSingleFlightForConcurrentReads) { + std::unique_ptr dir; + auto gated = std::make_shared(OpenTestFile(&dir)); + FileBlockCache cache(gated, sizeof(kContent) - 1, kBlockSize, /*capacity=*/1024, + GetDefaultPool()); + + // The first reader publishes block 0 = [18, 26) and blocks on its held fetch. + std::thread first([&cache]() { + std::string dest(4, 'X'); + EXPECT_TRUE(cache.Read({18, 4}, dest.data())); + EXPECT_EQ("stuv", std::string_view(dest.data(), 4)); + }); + while (gated->AsyncReadCount() == 0) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + + // The second reader finds the published block and waits for the same fetch. + std::thread second([&cache]() { + std::string dest(4, 'X'); + EXPECT_TRUE(cache.Read({22, 4}, dest.data())); + EXPECT_EQ("wxyz", std::string_view(dest.data(), 4)); + }); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + ASSERT_EQ(gated->AsyncReadCount(), 1); + gated->ReleaseAll(); + first.join(); + second.join(); + + ASSERT_EQ(gated->AsyncReadCount(), 1); + const FileBlockCache::Counters counters = cache.GetCounters(); + ASSERT_EQ(counters.fetches, 1u); + ASSERT_EQ(counters.hits, 2u); +} + +// Blocks are never evicted, so a read whose block does not fit into the capacity +// is declined instead of replacing a cached block. +TEST(TestFileBlockCache, TestExhaustedCapacityDeclinesNewBlocks) { + std::unique_ptr dir; + // The capacity holds exactly one block. + FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize, + /*capacity=*/kBlockSize, GetDefaultPool()); + + AssertServed({18, 4}, "stuv", &cache); + // Block 1 = [10, 18) does not fit anymore. + AssertDeclined({10, 4}, &cache); + // The block already cached still serves its reads. + AssertServed({20, 2}, "uv", &cache); + + const FileBlockCache::Counters counters = cache.GetCounters(); + ASSERT_EQ(counters.fetches, 1u); + ASSERT_EQ(counters.fetch_bytes, 8u); + ASSERT_EQ(counters.hits, 2u); +} + +// Reads that one block cannot serve are declined: a read straddling two blocks, +// a read larger than a block and a read reaching past EOF. +TEST(TestFileBlockCache, TestDeclinesStraddlingAndOversizedReads) { + std::unique_ptr dir; + FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize, + /*capacity=*/1024, GetDefaultPool()); + + // [16, 20) straddles block 1 = [10, 18) and block 0 = [18, 26). + AssertDeclined({16, 4}, &cache); + // Larger than one block. + AssertDeclined({0, 12}, &cache); + // Reaches past the end of the file. + AssertDeclined({24, 4}, &cache); + // Starts past the end of the file. + AssertDeclined({26, 2}, &cache); + + const FileBlockCache::Counters counters = cache.GetCounters(); + ASSERT_EQ(counters.fetches, 0u); + ASSERT_EQ(counters.hits, 0u); +} + +// A zero capacity or a zero block size turns the cache off entirely. +TEST(TestFileBlockCache, TestDisabledByZeroCapacityOrBlockSize) { + std::unique_ptr dir; + std::shared_ptr in = OpenTestFile(&dir); + + FileBlockCache no_capacity(in, sizeof(kContent) - 1, kBlockSize, /*capacity=*/0, + GetDefaultPool()); + AssertDeclined({18, 4}, &no_capacity); + ASSERT_EQ(no_capacity.GetCounters().fetches, 0u); + + FileBlockCache no_block_size(in, sizeof(kContent) - 1, /*block_size=*/0, /*capacity=*/1024, + GetDefaultPool()); + AssertDeclined({18, 4}, &no_block_size); + ASSERT_EQ(no_block_size.GetCounters().fetches, 0u); +} + +// ResetCounters() keeps the cached blocks, which cache the file rather than a +// round of reads, while Release() drops them and makes the next read fetch again. +TEST(TestFileBlockCache, TestResetCountersKeepsBlocksAndReleaseDropsThem) { + std::unique_ptr dir; + FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize, + /*capacity=*/1024, GetDefaultPool()); + + AssertServed({18, 4}, "stuv", &cache); + ASSERT_EQ(cache.GetCounters().fetches, 1u); + + cache.ResetCounters(); + ASSERT_EQ(cache.GetCounters().hits, 0u); + ASSERT_EQ(cache.GetCounters().fetches, 0u); + // The block is still there, so this read needs no fetch. + AssertServed({18, 4}, "stuv", &cache); + ASSERT_EQ(cache.GetCounters().hits, 1u); + ASSERT_EQ(cache.GetCounters().fetches, 0u); + + cache.Release(); + AssertServed({18, 4}, "stuv", &cache); + ASSERT_EQ(cache.GetCounters().hits, 2u); + ASSERT_EQ(cache.GetCounters().fetches, 1u); +} + +// A failed fetch is not reported to the caller, which reads its own bytes +// instead, and the failed block is not fetched again. +TEST(TestFileBlockCache, TestFetchErrorDeclinesTheReads) { + std::unique_ptr dir; + FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize, + /*capacity=*/1024, GetDefaultPool()); + auto io_hook = paimon::IOHook::GetInstance(); + paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); }); + io_hook->Reset(0, paimon::IOHook::Mode::RETURN_ERROR); + + AssertDeclined({18, 4}, &cache); + // The failed block is kept, so the later reads of it are declined without a + // second fetch, even once the reads would succeed again. + io_hook->Clear(); + AssertDeclined({20, 2}, &cache); + // The other blocks are unaffected. + AssertServed({10, 4}, "klmn", &cache); + + const FileBlockCache::Counters counters = cache.GetCounters(); + ASSERT_EQ(counters.fetches, 2u); + ASSERT_EQ(counters.hits, 1u); +} + +// A file size larger than the physical file - the size recorded by the file +// metadata is not necessarily the size of the file - makes the fetch of the +// block reaching past the end of the file fail. That must cost no more than the +// caching of that block: the reads it would serve are declined and read by the +// caller itself. +TEST(TestFileBlockCache, TestFileSizeLargerThanTheFileOnlyLosesTheLastBlock) { + std::unique_ptr dir; + // 30 instead of 26, so block 0 = [22, 30) has 4 bytes the file does not have. + FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1 + 4, kBlockSize, + /*capacity=*/1024, GetDefaultPool()); + + AssertDeclined({22, 4}, &cache); + // No second fetch of the block that cannot be read. + AssertDeclined({23, 2}, &cache); + // The blocks that the file does have still serve their reads. + AssertServed({10, 4}, "klmn", &cache); + + const FileBlockCache::Counters counters = cache.GetCounters(); + ASSERT_EQ(counters.fetches, 2u); + ASSERT_EQ(counters.hits, 1u); +} + +} // namespace paimon::test diff --git a/src/paimon/common/utils/read_ahead_cache.cpp b/src/paimon/common/utils/read_ahead_cache.cpp index a74b35202..b39d51310 100644 --- a/src/paimon/common/utils/read_ahead_cache.cpp +++ b/src/paimon/common/utils/read_ahead_cache.cpp @@ -30,6 +30,7 @@ #include #include "paimon/common/utils/byte_range_combiner.h" +#include "paimon/common/utils/file_block_cache.h" #include "paimon/common/utils/math.h" #include "paimon/memory/bytes.h" #include "paimon/metrics.h" @@ -79,29 +80,9 @@ void CopyRangeFromEntries(const std::vector& covering, const By } // namespace -CacheConfig::CacheConfig(uint64_t range_size_limit, uint64_t hole_size_limit, - uint64_t pre_buffer_limit) - : range_size_limit_(range_size_limit), - hole_size_limit_(hole_size_limit), - pre_buffer_limit_(pre_buffer_limit) {} - -CacheConfig::CacheConfig() - // Aligned with the reader's request granularity and with realistic data - // file sizes: - // - range_size_limit matches the parquet reader's 32 MiB request blocks - // (Arrow ReadRangeCache's own range limit); a smaller limit cuts entries - // below the request size, so a request can never be served from one piece. - // - pre_buffer_limit must exceed the LARGEST single read a reader issues - // (coalesced column-chunk reads of ~128 MiB were observed): fetches are - // only dispatched up to this window, so a request reaching past it can - // never be served and falls back to a second fetch of the same bytes. - : CacheConfig(/*range_size_limit=*/32 * 1024 * 1024, - /*hole_size_limit=*/8 * 1024, - /*pre_buffer_limit=*/256 * 1024 * 1024) {} - class ReadAheadCache::Impl { public: - Impl(const std::shared_ptr& stream, const CacheConfig& config, + Impl(const std::shared_ptr& stream, const CacheConfig& config, uint64_t file_size, const std::shared_ptr& memory_pool); ~Impl(); @@ -138,6 +119,10 @@ class ReadAheadCache::Impl { /// re-fetching the same bytes. void Cache(std::vector pending_indices); + /// Clear the prefetch state, waiting for the fetches still writing into the + /// entry buffers. Leaves the block cache untouched. + void ReleasePrefetchBuffers(); + std::shared_ptr stream_; CacheConfig config_; // Ordered by offset (so as to find a matching region by binary search) @@ -147,6 +132,9 @@ class ReadAheadCache::Impl { std::vector> is_cached_; std::vector pending_ranges_; bool is_initialized_ = false; + // Caches the reads that no registered range covers, or null when the block + // cache is disabled. Owns its own locking and counters. + std::unique_ptr block_cache_; // Statistics of the Read() requests issued to the cache, aggregated over // all streams sharing this cache. std::atomic read_count_{0}; @@ -242,18 +230,26 @@ void ReadAheadCache::Impl::PreBuffer(uint64_t offset) { } ReadAheadCache::Impl::Impl(const std::shared_ptr& stream, const CacheConfig& config, - const std::shared_ptr& memory_pool) - : stream_(stream), config_(config), memory_pool_(memory_pool) {} + uint64_t file_size, const std::shared_ptr& memory_pool) + : stream_(stream), config_(config), memory_pool_(memory_pool) { + // An unknown file size cannot be aligned to, and a zero limit or block size + // means the block cache is turned off: leave it null in those cases. + if (file_size > 0 && config_.GetBlockSize() > 0 && config_.GetBlockCacheLimit() > 0) { + block_cache_ = std::make_unique(stream, file_size, config_.GetBlockSize(), + config_.GetBlockCacheLimit(), memory_pool); + } +} ReadAheadCache::Impl::~Impl() { std::unique_lock lock(rw_mutex_); for (auto& entry : entries_) { entry.future.wait(); } + // The block cache waits for its own fetches when it is destroyed. } void ReadAheadCache::Impl::Reset() { - ReleaseBuffers(); + ReleasePrefetchBuffers(); read_count_.store(0, std::memory_order_relaxed); read_bytes_.store(0, std::memory_order_relaxed); hits_.store(0, std::memory_order_relaxed); @@ -262,9 +258,21 @@ void ReadAheadCache::Impl::Reset() { miss_bytes_.store(0, std::memory_order_relaxed); io_count_.store(0, std::memory_order_relaxed); io_bytes_.store(0, std::memory_order_relaxed); + if (block_cache_ != nullptr) { + // Only the counters: the blocks cache the file, not the registered + // ranges, and a reader resetting the cache reads the same file again. + block_cache_->ResetCounters(); + } } void ReadAheadCache::Impl::ReleaseBuffers() { + ReleasePrefetchBuffers(); + if (block_cache_ != nullptr) { + block_cache_->Release(); + } +} + +void ReadAheadCache::Impl::ReleasePrefetchBuffers() { std::unique_lock lock(rw_mutex_); // Entries are never evicted, so waiting on entries_ covers every // dispatched fetch: no async callback can outlive the stream or the @@ -293,8 +301,18 @@ void ReadAheadCache::Impl::CollectMetrics(std::shared_ptr* metrics) con m->SetCounter(ReadAheadCacheMetrics::READ_MISSES, misses_.load(std::memory_order_relaxed)); m->SetCounter(ReadAheadCacheMetrics::READ_MISS_BYTES, miss_bytes_.load(std::memory_order_relaxed)); - m->SetCounter(ReadAheadCacheMetrics::IO_COUNT, io_count_.load(std::memory_order_relaxed)); - m->SetCounter(ReadAheadCacheMetrics::IO_BYTES, io_bytes_.load(std::memory_order_relaxed)); + // The block cache keeps its own counters. Its fetches also go to the + // underlying stream, so they are part of the io counters too. + const FileBlockCache::Counters blocks = + block_cache_ != nullptr ? block_cache_->GetCounters() : FileBlockCache::Counters{}; + m->SetCounter(ReadAheadCacheMetrics::BLOCK_HITS, blocks.hits); + m->SetCounter(ReadAheadCacheMetrics::BLOCK_HIT_BYTES, blocks.hit_bytes); + m->SetCounter(ReadAheadCacheMetrics::BLOCK_FETCHES, blocks.fetches); + m->SetCounter(ReadAheadCacheMetrics::BLOCK_FETCH_BYTES, blocks.fetch_bytes); + m->SetCounter(ReadAheadCacheMetrics::IO_COUNT, + io_count_.load(std::memory_order_relaxed) + blocks.fetches); + m->SetCounter(ReadAheadCacheMetrics::IO_BYTES, + io_bytes_.load(std::memory_order_relaxed) + blocks.fetch_bytes); } void ReadAheadCache::Impl::Warmup() { @@ -350,6 +368,13 @@ Result ReadAheadCache::Impl::Read(const ByteRange& range, char* dest) { PreBuffer(range.offset); std::vector covering = FindCoveringEntries(range); if (covering.empty()) { + // No registered range covers this read: the block cache can still serve + // it, and then serve the readers of the other streams sharing this cache + // that are about to read the same bytes. + if (block_cache_ != nullptr && block_cache_->Read(range, dest)) { + // The block cache counts its own hits, see CollectMetrics(). + return true; + } CountMiss(range.length); return false; } @@ -379,9 +404,9 @@ void ReadAheadCache::Impl::DispatchFetches(const std::vector& fetc } ReadAheadCache::ReadAheadCache(const std::shared_ptr& stream, - const CacheConfig& config, + const CacheConfig& config, uint64_t file_size, const std::shared_ptr& memory_pool) - : impl_(std::make_unique(stream, config, memory_pool)) {} + : impl_(std::make_unique(stream, config, file_size, memory_pool)) {} ReadAheadCache::~ReadAheadCache() = default; diff --git a/src/paimon/common/utils/read_ahead_cache.h b/src/paimon/common/utils/read_ahead_cache.h index f039c22e8..d92520197 100644 --- a/src/paimon/common/utils/read_ahead_cache.h +++ b/src/paimon/common/utils/read_ahead_cache.h @@ -48,6 +48,16 @@ class PAIMON_EXPORT ReadAheadCacheMetrics { static inline const char READ_HIT_BYTES[] = "read-ahead-cache.read.hit-bytes"; static inline const char READ_MISSES[] = "read-ahead-cache.read.misses"; static inline const char READ_MISS_BYTES[] = "read-ahead-cache.read.miss-bytes"; + /// Number of Read() requests served by the block cache, and the bytes they + /// copied out of it. A read is counted either as a hit, a block hit or a + /// miss, so `read.count = read.hits + block.hits + read.misses`. + static inline const char BLOCK_HITS[] = "read-ahead-cache.block.hits"; + static inline const char BLOCK_HIT_BYTES[] = "read-ahead-cache.block.hit-bytes"; + /// Block fetches issued to the underlying stream, and their bytes. Both are + /// a subset of the io counters below, so comparing them tells how many bytes + /// the block granularity added on top of the requested ones. + static inline const char BLOCK_FETCHES[] = "read-ahead-cache.block.fetches"; + static inline const char BLOCK_FETCH_BYTES[] = "read-ahead-cache.block.fetch-bytes"; /// Number of prefetch IO requests actually issued to the underlying stream. static inline const char IO_COUNT[] = "read-ahead-cache.io.count"; /// Total bytes requested by the prefetch IOs issued to the underlying stream. @@ -85,11 +95,24 @@ struct PAIMON_EXPORT ByteRange { /// The cache never evicts: every published range stays cached until /// ReleaseBuffers() or Reset(). It is meant to hold the prefetched ranges of /// a single data file, whose size is bounded by the reader's scan scope. +/// +/// Reads that the prefetched ranges do not cover (the footer and the page index +/// of a parquet file are read before any range is registered) are served by a +/// FileBlockCache instead of being left to the caller. That block cache is owned +/// by this one and shares its lifetime: it is configured from `block_size` and +/// `block_cache_limit`, it survives Reset() - the blocks belong to the file +/// rather than to a registration round - and it is released by ReleaseBuffers(). class PAIMON_EXPORT ReadAheadCache { public: /// Construct a read cache with given options + /// @param stream The stream the cache fetches from. + /// @param config The cache configuration. + /// @param file_size Size of the file behind `stream`, used to align the + /// block cache to the end of the file. Zero means unknown and disables the + /// block cache. + /// @param memory_pool The pool the cached buffers are allocated from. ReadAheadCache(const std::shared_ptr& stream, const CacheConfig& config, - const std::shared_ptr& memory_pool); + uint64_t file_size, const std::shared_ptr& memory_pool); ~ReadAheadCache(); /// Initialize the cache with given byte ranges to be cached. @@ -104,9 +127,12 @@ class PAIMON_EXPORT ReadAheadCache { /// /// Multi-segment hits are copied into `dest` segment by segment, without /// an intermediate assembled buffer. + /// + /// A range that no registered range covers may still be served by the block + /// cache, see the class documentation. /// @param range The byte range to read. /// @param dest Destination buffer with at least `range.length` bytes. - /// @return true if the range was served from the cache and `dest` was + /// @return true if the range was served by the cache and `dest` was /// filled; false on cache miss (`dest` is left untouched). Result Read(const ByteRange& range, char* dest); @@ -129,6 +155,9 @@ class PAIMON_EXPORT ReadAheadCache { /// This method waits for all ongoing asynchronous read operations to complete, /// clears all cached entries, and resets the internal state so that Init() can be called again. /// After calling Reset, the cache can be safely re-initialized with new ranges. + /// + /// The block cache is kept: it caches the file rather than the registered + /// ranges, and a reader reusing the cache reads the same file again. void Reset(); /// Release all cached buffers and pending ranges while keeping the hit/miss @@ -136,7 +165,8 @@ class PAIMON_EXPORT ReadAheadCache { /// /// Unlike Reset(), the counters recorded by Read() remain readable through /// CollectMetrics() afterwards, so this is safe to call when the owning reader - /// is closed while its metrics are still being aggregated. + /// is closed while its metrics are still being aggregated. The block cache is + /// released too, as the file is not read again. void ReleaseBuffers(); private: diff --git a/src/paimon/common/utils/read_ahead_cache_test.cpp b/src/paimon/common/utils/read_ahead_cache_test.cpp index e7900b7b6..065b4131c 100644 --- a/src/paimon/common/utils/read_ahead_cache_test.cpp +++ b/src/paimon/common/utils/read_ahead_cache_test.cpp @@ -21,7 +21,6 @@ #include #include -#include #include #include @@ -31,21 +30,32 @@ #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 { -// Helper to create a test file, write content, and return a ready ReadAheadCache. -struct TestCacheEnv { - std::string path; - std::shared_ptr cache; - std::shared_ptr pool; -}; +// The range limits the tests exercise. The defaults are sized for real data +// files, which the small test files would never reach. +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; +} -TestCacheEnv CreateTestFileAndCache(const std::string& filename, const std::string& content, - const paimon::CacheConfig& config, - std::vector ranges) { - auto dir = UniqueTestDirectory::Create(); +// 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 CreateTestFileAndCache(const std::string& filename, + const std::string& content, + const CacheConfig& config, + std::vector ranges, + uint64_t file_size = 0) { + std::unique_ptr dir = UniqueTestDirectory::Create(); EXPECT_TRUE(dir); std::string path = dir->Str() + "/" + filename; std::ofstream file(path, std::ios::binary); @@ -54,17 +64,13 @@ TestCacheEnv CreateTestFileAndCache(const std::string& filename, const std::stri 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(); + EXPECT_OK_AND_ASSIGN(std::unique_ptr fs, FileSystemFactory::Get("local", path, {})); + EXPECT_OK_AND_ASSIGN(std::unique_ptr in, fs->Open(path)); - auto pool = GetDefaultPool(); - auto cache = std::make_shared(std::move(in), config, pool); + std::shared_ptr cache = + std::make_shared(std::move(in), config, file_size, GetDefaultPool()); EXPECT_OK(cache->Init(std::move(ranges))); - return {path, cache, pool}; + return cache; } // Assert that reading the range is a cache hit filling the destination with @@ -74,7 +80,7 @@ void AssertReadEquals(const ByteRange& range, const std::string& expected, ReadA 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_EQ(expected, std::string_view(dest.data(), range.length)); } // Assert that reading the range misses and leaves the destination untouched. @@ -83,84 +89,18 @@ void AssertReadMiss(const ByteRange& range, ReadAheadCache* cache) { bool hit = true; ASSERT_OK_AND_ASSIGN(hit, cache->Read(range, dest.data())); ASSERT_FALSE(hit); - EXPECT_EQ(std::string(dest.size(), 'X'), dest); + ASSERT_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 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 GetPos() const override { - return inner_->GetPos(); - } - Result Read(char* buffer, int64_t size) override { - return inner_->Read(buffer, size); - } - Result 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&& callback) override { - std::lock_guard lock(mutex_); - async_read_count_++; - pending_.push_back({buffer, size, offset, std::move(callback)}); - } - Result GetUri() const override { - return inner_->GetUri(); - } - Result Length() const override { - return inner_->Length(); - } - - int AsyncReadCount() { - std::lock_guard lock(mutex_); - return async_read_count_; - } - - /// Complete all held fetches against the underlying stream. - void ReleaseAll() { - std::vector taken; - { - std::lock_guard lock(mutex_); - taken = std::move(pending_); - pending_.clear(); - } - for (auto& read : taken) { - Result 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 callback; - }; - - std::shared_ptr inner_; - std::mutex mutex_; - std::vector 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); + CacheConfig config = + TestCacheConfig(/*range_size_limit=*/10, + /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024 * 1024); std::string content = "abcdefghijklmnopqrstuvwxyz"; - auto env = CreateTestFileAndCache( + std::shared_ptr cache_ptr = 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; + ReadAheadCache& cache = *cache_ptr; AssertReadEquals({20, 2}, "uv", &cache); AssertReadEquals({1, 2}, "bc", &cache); @@ -183,13 +123,14 @@ TEST(TestReadAheadCache, TestBasics) { // 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); + 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 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; + std::shared_ptr cache_ptr = + CreateTestFileAndCache("data_file", content, config, {{0, 25}}); + ReadAheadCache& cache = *cache_ptr; // Spans all three entries. AssertReadEquals({5, 20}, "fghijklmnopqrstuvwxy", &cache); @@ -217,11 +158,12 @@ TEST(TestReadAheadCache, TestMultiSegmentContiguousHit) { // 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); + CacheConfig config = TestCacheConfig(/*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; + std::shared_ptr 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); @@ -230,11 +172,12 @@ TEST(TestReadAheadCache, TestRepeatedReadCacheReuse) { // 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); + CacheConfig config = TestCacheConfig(/*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; + std::shared_ptr cache_ptr = + CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}, {16, 5}}); + ReadAheadCache& cache = *cache_ptr; AssertReadEquals({0, 5}, "abcde", &cache); @@ -246,11 +189,13 @@ TEST(TestReadAheadCache, TestNoEvictionKeepsAllRanges) { // 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); + CacheConfig config = + TestCacheConfig(/*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; + std::shared_ptr 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. @@ -286,11 +231,13 @@ TEST(TestReadAheadCache, TestMetrics) { // 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); + CacheConfig config = + TestCacheConfig(/*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; + std::shared_ptr cache_ptr = + CreateTestFileAndCache("data_file", content, config, {{0, 5}}); + ReadAheadCache& cache = *cache_ptr; AssertReadEquals({0, 5}, "abcde", &cache); @@ -341,28 +288,30 @@ TEST(TestReadAheadCache, TestReleaseBuffersKeepsMetrics) { // 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); + 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. { - auto env = CreateTestFileAndCache("data_file", content, config, {{0, 10}}); + std::shared_ptr 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(env.cache->Read({0, 5}, dest.data()), + 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. { - auto env = CreateTestFileAndCache("data_file", content, config, {{0, 25}}); + std::shared_ptr 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(env.cache->Read({0, 20}, dest.data()), + ASSERT_NOK_WITH_MSG(cache_ptr->Read({0, 20}, dest.data()), "io hook triggered io error at position"); } } @@ -371,42 +320,45 @@ TEST(TestReadAheadCache, TestPrefetchIOErrorPropagation) { // 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); + CacheConfig config = TestCacheConfig(/*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}}); + std::shared_ptr cache1 = + CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5}}); + cache1->Warmup(); + std::shared_ptr 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", env1.cache.get()); - AssertReadEquals({8, 5}, "ijklm", env1.cache.get()); + 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(env2.cache->Read({0, 5}, dest.data())); + ASSERT_NOK(cache2->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); + CacheConfig config = TestCacheConfig(/*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()); + std::shared_ptr 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(/*range_size_limit=*/10, - /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); + CacheConfig config = TestCacheConfig(/*range_size_limit=*/10, + /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); std::string content = "abcdefghijklmnopqrstuvwxyz"; auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); @@ -416,11 +368,11 @@ TEST(TestReadAheadCache, TestInFlightEntryServesRacingReader) { 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::unique_ptr fs, FileSystemFactory::Get("local", path, {})); ASSERT_OK_AND_ASSIGN(std::shared_ptr in, fs->Open(path)); auto gated = std::make_shared(std::move(in)); - ReadAheadCache cache(gated, config, GetDefaultPool()); + ReadAheadCache cache(gated, config, /*file_size=*/0, GetDefaultPool()); ASSERT_OK(cache.Init({{0, 5}})); cache.Warmup(); @@ -456,11 +408,12 @@ TEST(TestReadAheadCache, TestInFlightEntryServesRacingReader) { // 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); + CacheConfig config = TestCacheConfig(/*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; + std::shared_ptr 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(); }); @@ -483,11 +436,12 @@ TEST(TestReadAheadCache, TestPreBufferWindowLimit) { // 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); + CacheConfig config = TestCacheConfig(/*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; + std::shared_ptr cache_ptr = + CreateTestFileAndCache("data_file", content, config, {{0, 5}}); + ReadAheadCache& cache = *cache_ptr; Status status = cache.Init({{8, 5}}); ASSERT_FALSE(status.ok()); @@ -498,11 +452,12 @@ TEST(TestReadAheadCache, TestDoubleInit) { // 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); + CacheConfig config = TestCacheConfig(/*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; + std::shared_ptr cache_ptr = + CreateTestFileAndCache("data_file", content, config, {{0, 5}}); + ReadAheadCache& cache = *cache_ptr; AssertReadEquals({0, 5}, "abcde", &cache); @@ -517,25 +472,186 @@ TEST(TestReadAheadCache, TestReinitAfterReset) { // 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); + 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}. - auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5}, {6, 5}}); - auto& cache = *env.cache; + std::shared_ptr 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(/*range_size_limit=*/10, - /*hole_size_limit=*/2, /*pre_buffer_limit=*/1024); + CacheConfig config = TestCacheConfig(/*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 cache_ptr = + CreateTestFileAndCache("data_file", content, config, {{0, 5}}); + cache_ptr->CollectMetrics(/*metrics=*/nullptr); std::shared_ptr null_metrics; - env.cache->CollectMetrics(&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 = std::make_shared(); + 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 = std::make_shared(); + 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 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 = std::make_shared(); + 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 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 = std::make_shared(); + 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 a registration round: 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 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 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 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 diff --git a/src/paimon/core/operation/read_context_test.cpp b/src/paimon/core/operation/read_context_test.cpp index 1174568f3..3d362352e 100644 --- a/src/paimon/core/operation/read_context_test.cpp +++ b/src/paimon/core/operation/read_context_test.cpp @@ -59,8 +59,10 @@ TEST(ReadContextTest, TestSetContent) { ReadContextBuilder builder("table_root_path"); std::shared_ptr memory_pool = GetDefaultPool(); std::shared_ptr executor = CreateDefaultExecutor(); - CacheConfig cache_config(/*range_size_limit=*/512, /*hole_size_limit=*/128, - /*pre_buffer_limit=*/2048); + CacheConfig cache_config; + cache_config.SetRangeSizeLimit(512); + cache_config.SetHoleSizeLimit(128); + cache_config.SetPreBufferLimit(2048); builder.AddOption("key", "value"); builder.SetReadFieldNames({"f1", "f2"}); diff --git a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp index 3e4824e34..9cd7a68dd 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp +++ b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -1679,7 +1680,10 @@ TEST_F(ParquetFileBatchReaderTest, TestPreBufferRangeFeedsReadAheadCache) { /*enable_dictionary=*/true, /*max_row_group_length=*/10); ASSERT_OK_AND_ASSIGN(std::shared_ptr cache_stream, fs_->Open(file_path_)); - auto cache = std::make_shared(cache_stream, CacheConfig(), GetDefaultPool()); + // The file size is left unknown so the block cache stays off: this test is + // about the pre-buffered ranges feeding the cache. + auto cache = std::make_shared(cache_stream, CacheConfig(), /*file_size=*/0, + GetDefaultPool()); ASSERT_OK_AND_ASSIGN(std::unique_ptr reader_stream, fs_->Open(file_path_)); auto cache_input_stream = std::make_shared(std::move(reader_stream), cache); @@ -2038,4 +2042,128 @@ TEST_F(ParquetFileBatchReaderTest, TestDictionaryPassthroughRequiresEveryRowGrou reader->Close(); } +// Counts the positional reads landing in the tail region of the file, i.e. the +// footer and the page index of a parquet file. The counter is shared by all the +// streams of one test, so it totals the reads reaching the storage. +class TailReadCountingInputStream : public InputStream { + public: + TailReadCountingInputStream(std::unique_ptr input, int64_t tail_begin, + std::shared_ptr> tail_read_count) + : input_(std::move(input)), + tail_begin_(tail_begin), + tail_read_count_(std::move(tail_read_count)) {} + + Status Seek(int64_t offset, SeekOrigin origin) override { + return input_->Seek(offset, origin); + } + Result GetPos() const override { + return input_->GetPos(); + } + Result Read(char* buffer, int64_t size) override { + return input_->Read(buffer, size); + } + Result Read(char* buffer, int64_t size, int64_t offset) override { + CountIfInTail(offset); + return input_->Read(buffer, size, offset); + } + void ReadAsync(char* buffer, int64_t size, int64_t offset, + std::function&& callback) override { + CountIfInTail(offset); + return input_->ReadAsync(buffer, size, offset, std::move(callback)); + } + Status Close() override { + return input_->Close(); + } + Result GetUri() const override { + return input_->GetUri(); + } + Result Length() const override { + return input_->Length(); + } + + private: + void CountIfInTail(int64_t offset) { + if (offset >= tail_begin_) { + tail_read_count_->fetch_add(1); + } + } + + std::unique_ptr input_; + int64_t tail_begin_; + std::shared_ptr> tail_read_count_; +}; + +// The sub-readers of a prefetch reader share one ReadAheadCache and read the +// footer and the page index before any pre-buffer range is registered. The block +// cache turns all those tail reads into a single read of the last block. +TEST_F(ParquetFileBatchReaderTest, TestBlockCacheSharesTailReadsAcrossReaders) { + arrow::FieldVector fields = {arrow::field("f0", arrow::int32())}; + auto src_array = MakeSequentialIntData(60000); + auto arrow_schema = arrow::schema(fields); + WriteArray(file_path_, src_array, arrow_schema, /*write_batch_size=*/1024, + /*enable_dictionary=*/false, /*max_row_group_length=*/30000); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr length_stream, fs_->Open(file_path_)); + ASSERT_OK_AND_ASSIGN(int64_t file_length, length_stream->Length()); + ASSERT_OK(length_stream->Close()); + // The tail region is the last block of the cache, which is also the footer + // read size of the parquet reader. + const int64_t tail_begin = file_length - static_cast(CacheConfig().GetBlockSize()); + ASSERT_GT(tail_begin, 0); + + auto tail_read_count = std::make_shared>(0); + ASSERT_OK_AND_ASSIGN(std::unique_ptr cache_stream, fs_->Open(file_path_)); + auto counting_cache_stream = std::make_shared( + std::move(cache_stream), tail_begin, tail_read_count); + auto cache = std::make_shared( + counting_cache_stream, CacheConfig(), static_cast(file_length), GetDefaultPool()); + + std::map options; + options[PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] = "true"; + ParquetReaderBuilder builder(options, batch_size_); + builder.WithMemoryPool(GetDefaultPool()); + + constexpr int32_t kReaderCount = 3; + std::vector>>> futures; + for (int32_t i = 0; i < kReaderCount; ++i) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader_stream, fs_->Open(file_path_)); + auto counting = std::make_unique(std::move(reader_stream), + tail_begin, tail_read_count); + std::shared_ptr cache_input_stream = + std::make_shared(std::move(counting), cache); + futures.push_back(std::async(std::launch::async, [&builder, cache_input_stream]() { + return builder.Build(cache_input_stream); + })); + } + std::vector> readers; + for (auto& future : futures) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, future.get()); + readers.push_back(std::move(reader)); + } + + // Only the last 1000 rows match, so the page index of the last row group is + // read to filter its pages. + std::shared_ptr predicate = PredicateBuilder::GreaterOrEqual( + /*field_index=*/0, /*field_name=*/"f0", FieldType::INT, Literal(59000)); + for (auto& reader : readers) { + auto parquet_batch_reader = dynamic_cast(reader.get()); + ASSERT_TRUE(parquet_batch_reader); + std::unique_ptr c_schema = std::make_unique(); + ASSERT_TRUE(arrow::ExportSchema(*arrow_schema, c_schema.get()).ok()); + ASSERT_OK(parquet_batch_reader->SetReadSchema(c_schema.get(), predicate, std::nullopt)); + } + + // One block fetch served the footer read and the page index reads of all the + // readers. + ASSERT_EQ(tail_read_count->load(), 1); + std::shared_ptr metrics = std::make_shared(); + cache->CollectMetrics(&metrics); + ASSERT_OK_AND_ASSIGN(uint64_t block_fetches, + metrics->GetCounter(ReadAheadCacheMetrics::BLOCK_FETCHES)); + ASSERT_EQ(block_fetches, 1u); + ASSERT_OK_AND_ASSIGN(uint64_t block_hits, + metrics->GetCounter(ReadAheadCacheMetrics::BLOCK_HITS)); + ASSERT_GT(block_hits, 1u); +} + } // namespace paimon::parquet::test diff --git a/src/paimon/testing/utils/gated_async_input_stream.h b/src/paimon/testing/utils/gated_async_input_stream.h new file mode 100644 index 000000000..0f5acb66e --- /dev/null +++ b/src/paimon/testing/utils/gated_async_input_stream.h @@ -0,0 +1,115 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include "paimon/fs/file_system.h" + +namespace paimon::test { + +/// An InputStream wrapper that holds the ReadAsync callbacks until ReleaseAll() +/// is called, letting tests observe a cache while its fetches are in flight. +class GatedAsyncInputStream : public InputStream { + public: + explicit GatedAsyncInputStream(std::shared_ptr 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 GetPos() const override { + return inner_->GetPos(); + } + Result Read(char* buffer, int64_t size) override { + return inner_->Read(buffer, size); + } + Result 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&& callback) override { + std::lock_guard lock(mutex_); + async_read_count_++; + pending_.push_back({buffer, size, offset, std::move(callback)}); + } + Result GetUri() const override { + return inner_->GetUri(); + } + Result Length() const override { + return inner_->Length(); + } + + int32_t AsyncReadCount() const { + std::lock_guard lock(mutex_); + return async_read_count_; + } + + /// Complete all held fetches against the underlying stream. + void ReleaseAll() { + std::vector taken; + { + std::lock_guard lock(mutex_); + taken = std::move(pending_); + pending_.clear(); + } + for (auto& read : taken) { + Result res = inner_->Read(read.buffer, read.size, read.offset); + read.callback(res.ok() ? Status::OK() : res.status()); + } + } + + /// Fail all held fetches without touching the underlying stream, so that a + /// test can observe how a cache reports a failed fetch. + void FailAll(const Status& status) { + std::vector taken; + { + std::lock_guard lock(mutex_); + taken = std::move(pending_); + pending_.clear(); + } + for (auto& read : taken) { + read.callback(status); + } + } + + private: + struct PendingRead { + char* buffer; + int64_t size; + int64_t offset; + std::function callback; + }; + + std::shared_ptr inner_; + mutable std::mutex mutex_; + std::vector pending_; + int32_t async_read_count_ = 0; +}; + +} // namespace paimon::test