This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 6c69e996 feat(core): add a block cache for the reads the prefetched
ranges do not cover (#272)
6c69e996 is described below
commit 6c69e996e0f983918650f8317310c79f8ab1d5ca
Author: Yonghao Fang <[email protected]>
AuthorDate: Thu Sep 10 09:18:37 2026 +0800
feat(core): add a block cache for the reads the prefetched ranges do not
cover (#272)
---
include/paimon/utils/prefetch_cache_config.h | 57 ++-
src/paimon/CMakeLists.txt | 2 +
src/paimon/common/io/cache_input_stream_test.cpp | 25 +-
src/paimon/common/memory/bytes_utils.h | 60 +++
src/paimon/common/metrics/atomic_counter_pair.h | 57 +++
.../reader/prefetch_file_batch_reader_impl.cpp | 9 +-
.../prefetch_file_batch_reader_impl_test.cpp | 18 +-
src/paimon/common/utils/file_block_cache.cpp | 168 +++++++
src/paimon/common/utils/file_block_cache.h | 155 +++++++
src/paimon/common/utils/file_block_cache_test.cpp | 328 ++++++++++++++
src/paimon/common/utils/read_ahead_cache.cpp | 153 ++++---
src/paimon/common/utils/read_ahead_cache.h | 60 ++-
src/paimon/common/utils/read_ahead_cache_test.cpp | 494 ++++++++++++++-------
src/paimon/core/operation/read_context_test.cpp | 6 +-
.../parquet/parquet_file_batch_reader_test.cpp | 142 +++++-
.../testing/utils/gated_async_input_stream.h | 139 ++++++
16 files changed, 1598 insertions(+), 275 deletions(-)
diff --git a/include/paimon/utils/prefetch_cache_config.h
b/include/paimon/utils/prefetch_cache_config.h
index 4bbf1ecd..bd1197a6 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,51 @@ 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 is the granularity the reads no prefetched range covers are
+ // shared at: small enough that a metadata read at the tail of a file is
+ // served by one block instead of straddling two, large enough that a block
+ // fetch does not pull in much more than the reads ask for.
+ 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 6fc1deaa..b034b30b 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -181,6 +181,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
@@ -687,6 +688,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 0b14dc33..072af67b 100644
--- a/src/paimon/common/io/cache_input_stream_test.cpp
+++ b/src/paimon/common/io/cache_input_stream_test.cpp
@@ -40,7 +40,10 @@ namespace paimon::test {
class CacheInputStreamTest : public ::testing::Test {
public:
void SetUp() override {
- pool_ = GetDefaultPool();
+ // A pool of its own, so that a cache buffer outliving the pool it was
+ // allocated from shows up instead of being covered by the global pool,
+ // which never goes away.
+ pool_ = std::shared_ptr<MemoryPool>(GetMemoryPool());
test_dir_ = UniqueTestDirectory::Create();
ASSERT_TRUE(test_dir_);
content_ = "abcdefghijklmnopqrstuvwxyz0123456789";
@@ -60,9 +63,14 @@ class CacheInputStreamTest : public ::testing::Test {
std::shared_ptr<ReadAheadCache> CreateCache(std::vector<ByteRange> ranges)
{
auto stream = OpenFile();
- CacheConfig config(/*range_size_limit=*/1024,
- /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024 *
1024);
- auto cache = std::make_shared<ReadAheadCache>(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<ReadAheadCache>(std::move(stream), config,
/*file_size=*/0, pool_);
EXPECT_OK(cache->Init(std::move(ranges)));
return cache;
}
@@ -204,9 +212,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<ReadAheadCache>(std::move(cache_stream),
config, pool_);
+ CacheConfig config;
+ config.SetRangeSizeLimit(1024);
+ config.SetHoleSizeLimit(0);
+ config.SetPreBufferLimit(1024 * 1024);
+ auto cache = std::make_shared<ReadAheadCache>(std::move(cache_stream),
config,
+ /*file_size=*/0, pool_);
ASSERT_OK(cache->Init(std::vector<ByteRange>{{0, 10}}));
// Now activate IOHook so that the prefetch IO (triggered by
cache_->Read -> PreBuffer)
diff --git a/src/paimon/common/memory/bytes_utils.h
b/src/paimon/common/memory/bytes_utils.h
new file mode 100644
index 00000000..bfc1c1ca
--- /dev/null
+++ b/src/paimon/common/memory/bytes_utils.h
@@ -0,0 +1,60 @@
+/*
+ * 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 <cstddef>
+#include <memory>
+#include <utility>
+
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+
+namespace paimon {
+
+/// Allocate a shared buffer that keeps the pool it was allocated from alive.
+///
+/// A `Bytes` holds a raw pointer to its pool and frees its allocation through
+/// it, so it must not outlive the pool. Holding both in one owner is enough as
+/// long as that owner drops the buffer, but not once the buffer is handed to
an
+/// asynchronous read: the callback of the read owns a reference to the buffer,
+/// and a stream that destroys the callback after it has resolved the read -
the
+/// object store streams do - lets an IO thread drop the last reference to the
+/// buffer after the owner and the pool are already gone. Binding the pool to
+/// the buffer makes the buffer keep its own allocator alive, wherever its last
+/// reference is dropped.
+///
+/// @param size Number of bytes to allocate.
+/// @param pool Memory pool to allocate from, which the returned buffer keeps
+/// alive.
+inline std::shared_ptr<Bytes> AllocateBytesKeepingPoolAlive(
+ size_t size, const std::shared_ptr<MemoryPool>& pool) {
+ // The pool is declared before the buffer, so the holder destroys the
buffer
+ // first and the pool it was allocated from second.
+ struct BytesWithMemoryPool {
+ std::shared_ptr<MemoryPool> pool;
+ std::shared_ptr<Bytes> bytes;
+ };
+ auto holder = std::make_shared<BytesWithMemoryPool>(
+ BytesWithMemoryPool{pool, std::make_shared<Bytes>(size, pool.get())});
+ Bytes* bytes = holder->bytes.get();
+ return std::shared_ptr<Bytes>(std::move(holder), bytes);
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/metrics/atomic_counter_pair.h
b/src/paimon/common/metrics/atomic_counter_pair.h
new file mode 100644
index 00000000..5a8297ac
--- /dev/null
+++ b/src/paimon/common/metrics/atomic_counter_pair.h
@@ -0,0 +1,57 @@
+/*
+ * 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 <atomic>
+#include <cstdint>
+
+namespace paimon {
+
+/// How many requests of one kind were made and how many bytes they covered,
+/// which is the shape every counter of the read path has. Grouping the two
+/// keeps a counter and its byte counter from drifting apart.
+///
+/// The atomics are relaxed throughout: the counters are only reported, never
+/// used to order anything, and the read path increments them on every request.
+struct AtomicCounterPair {
+ std::atomic<uint64_t> count{0};
+ std::atomic<uint64_t> bytes{0};
+
+ /// Record one request covering `size` bytes.
+ void Add(uint64_t size) {
+ count.fetch_add(1, std::memory_order_relaxed);
+ bytes.fetch_add(size, std::memory_order_relaxed);
+ }
+
+ void Reset() {
+ count.store(0, std::memory_order_relaxed);
+ bytes.store(0, std::memory_order_relaxed);
+ }
+
+ uint64_t Count() const {
+ return count.load(std::memory_order_relaxed);
+ }
+
+ uint64_t Bytes() const {
+ return bytes.load(std::memory_order_relaxed);
+ }
+};
+
+} // namespace paimon
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 f24dd2b5..51589ba3 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<std::unique_ptr<PrefetchFileBatchReaderImpl>>
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<std::unique_ptr<PrefetchFileBatchReaderImpl>>
PrefetchFileBatchReaderImpl
if (io_metrics) {
input_stream = std::make_shared<MetricsInputStream>(input_stream,
io_metrics);
}
- cache = std::make_shared<ReadAheadCache>(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<ReadAheadCache>(input_stream, cache_config,
+
static_cast<uint64_t>(data_file_size), pool);
}
std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> 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 02f9ba8d..5e0c40d4 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 00000000..829ffb23
--- /dev/null
+++ b/src/paimon/common/utils/file_block_cache.cpp
@@ -0,0 +1,168 @@
+/*
+ * 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 <cstring>
+
+#include "paimon/common/memory/bytes_utils.h"
+
+namespace paimon {
+
+FileBlockCache::FileBlockCache(const std::shared_ptr<InputStream>& stream,
uint64_t file_size,
+ uint64_t block_size, uint64_t capacity,
+ const std::shared_ptr<MemoryPool>& 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 they read from.
+ std::lock_guard<std::mutex> 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> 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<std::mutex> 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>();
+ block->range = block_range;
+ // The buffer keeps the pool alive, for the fetch callbacks that a
+ // stream destroys later than it resolves them.
+ block->buffer = AllocateBytesKeepingPoolAlive(block_range.length,
memory_pool_);
+ block->promise = std::make_shared<std::promise<Status>>();
+ 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_.Add(range.length);
+ return true;
+}
+
+void FileBlockCache::Release() {
+ std::lock_guard<std::mutex> 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_.Reset();
+ fetches_.Reset();
+}
+
+FileBlockCache::Counters FileBlockCache::GetCounters() const {
+ Counters counters;
+ counters.hits = hits_.Count();
+ counters.hit_bytes = hits_.Bytes();
+ counters.fetches = fetches_.Count();
+ counters.fetch_bytes = fetches_.Bytes();
+ 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>& block) {
+ fetches_.Add(block->range.length);
+ 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. The buffer keeps the
memory
+ // pool alive as well, so a callback outliving this cache still frees the
+ // buffer against a live pool.
+ stream_->ReadAsync(buffer->data(), static_cast<int64_t>(buffer->size()),
+ static_cast<int64_t>(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 00000000..3a2d5737
--- /dev/null
+++ b/src/paimon/common/utils/file_block_cache.h
@@ -0,0 +1,155 @@
+/*
+ * 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 <cstdint>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <unordered_map>
+
+#include "paimon/common/metrics/atomic_counter_pair.h"
+#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, the fallback below the prefetched
+/// ranges: a read that no registered range covers is served here at block
+/// granularity instead of being left to the caller, which would read it from
the
+/// underlying stream uncached and pay for it again on the next reader. It is a
+/// general mechanism over the whole file - any reader, any offset, any round
of
+/// registered ranges - not a cache of one particular kind of read.
+///
+/// Blocks are aligned to the END of the file: block 0 is
+/// [file_size - block_size, file_size). The reads served here concentrate at
the
+/// tail of a file: a reader reads the metadata of its file before it knows
which
+/// ranges to register, and every reader of the file reads the same bytes. End
+/// alignment gives that region whole blocks - a read of the last block_size
bytes
+/// of a file is one block whatever the file size - and leaves the only partial
+/// block at the head of the file. 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().
+///
+/// One block serves one read: a read longer than a block, or straddling two,
is
+/// declined and left to the caller instead of being fetched in pieces, so a
+/// served read stays one block, one fetch and one copy.
+///
+/// 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. A
snapshot:
+ /// the counters keep being recorded while it is read.
+ 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<InputStream>& stream, uint64_t
file_size,
+ uint64_t block_size, uint64_t capacity,
+ const std::shared_ptr<MemoryPool>& 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<Bytes> buffer;
+ std::shared_ptr<std::promise<Status>> promise;
+ // shared_future, as every reader of the block waits on it.
+ std::shared_future<Status> 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>& block);
+
+ std::shared_ptr<InputStream> stream_;
+ uint64_t file_size_;
+ uint64_t block_size_;
+ uint64_t capacity_;
+ std::shared_ptr<MemoryPool> 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<uint64_t, std::shared_ptr<Block>> blocks_;
+ // Bytes held by blocks_, guarded by mutex_ and bounded by capacity_.
+ uint64_t cached_bytes_ = 0;
+ // The requests served out of a block, and the block fetches issued to the
+ // underlying stream.
+ AtomicCounterPair hits_;
+ AtomicCounterPair fetches_;
+};
+
+} // 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 00000000..aadcc404
--- /dev/null
+++ b/src/paimon/common/utils/file_block_cache_test.cpp
@@ -0,0 +1,328 @@
+/*
+ * 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 <chrono>
+#include <fstream>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <thread>
+#include <utility>
+
+#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;
+
+// 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 test content into a fresh directory and open it for reading. The
+// directory is returned so that it outlives the stream.
+std::shared_ptr<InputStream>
OpenTestFile(std::unique_ptr<UniqueTestDirectory>* 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<std::unique_ptr<FileSystem>> fs = FileSystemFactory::Get("local",
path, {});
+ EXPECT_OK(fs.status());
+ Result<std::unique_ptr<InputStream>> 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<UniqueTestDirectory> dir;
+ FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize,
+ /*capacity=*/1024, TestPool());
+
+ // 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<UniqueTestDirectory> dir;
+ FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize,
+ /*capacity=*/1024, TestPool());
+
+ 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<UniqueTestDirectory> dir;
+ auto gated = std::make_shared<GatedAsyncInputStream>(OpenTestFile(&dir));
+ FileBlockCache cache(gated, sizeof(kContent) - 1, kBlockSize,
/*capacity=*/1024, TestPool());
+
+ // 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<UniqueTestDirectory> dir;
+ // The capacity holds exactly one block.
+ FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize,
+ /*capacity=*/kBlockSize, TestPool());
+
+ 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<UniqueTestDirectory> dir;
+ FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize,
+ /*capacity=*/1024, TestPool());
+
+ // [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<UniqueTestDirectory> dir;
+ std::shared_ptr<InputStream> in = OpenTestFile(&dir);
+
+ FileBlockCache no_capacity(in, sizeof(kContent) - 1, kBlockSize,
/*capacity=*/0, TestPool());
+ 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,
+ TestPool());
+ 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<UniqueTestDirectory> dir;
+ FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize,
+ /*capacity=*/1024, TestPool());
+
+ 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<UniqueTestDirectory> dir;
+ FileBlockCache cache(OpenTestFile(&dir), sizeof(kContent) - 1, kBlockSize,
+ /*capacity=*/1024, TestPool());
+ 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<UniqueTestDirectory> 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, TestPool());
+
+ 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);
+}
+
+// An object store stream destroys the callback of a read after it has resolved
+// it, so the last reference to a block 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(TestFileBlockCache, TestBlockBufferKeepsPoolAliveAfterCacheIsGone) {
+ std::unique_ptr<UniqueTestDirectory> dir;
+ auto gated = std::make_shared<GatedAsyncInputStream>(OpenTestFile(&dir));
+ 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.
+ FileBlockCache cache(gated, sizeof(kContent) - 1, kBlockSize,
/*capacity=*/1024, pool);
+
+ std::thread reader([&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 fetch is completed, but the stream keeps its callback - and
with it
+ // a reference to the block buffer - alive.
+ gated->ReleaseAllKeepingCallbacks();
+ reader.join();
+ }
+
+ // 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());
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/utils/read_ahead_cache.cpp
b/src/paimon/common/utils/read_ahead_cache.cpp
index a74b3520..6e517aaa 100644
--- a/src/paimon/common/utils/read_ahead_cache.cpp
+++ b/src/paimon/common/utils/read_ahead_cache.cpp
@@ -29,7 +29,10 @@
#include <future>
#include <shared_mutex>
+#include "paimon/common/memory/bytes_utils.h"
+#include "paimon/common/metrics/atomic_counter_pair.h"
#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 +82,9 @@ void CopyRangeFromEntries(const
std::vector<RangeCacheEntry>& 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<InputStream>& stream, const CacheConfig& config,
+ Impl(const std::shared_ptr<InputStream>& stream, const CacheConfig&
config, uint64_t file_size,
const std::shared_ptr<MemoryPool>& memory_pool);
~Impl();
@@ -121,14 +104,6 @@ class ReadAheadCache::Impl {
/// so the caller may use them after releasing the lock.
std::vector<RangeCacheEntry> FindCoveringEntries(const ByteRange& range);
void PreBuffer(uint64_t offset);
- void CountHit(uint64_t size) {
- hits_.fetch_add(1, std::memory_order_relaxed);
- hit_bytes_.fetch_add(size, std::memory_order_relaxed);
- }
- void CountMiss(uint64_t size) {
- misses_.fetch_add(1, std::memory_order_relaxed);
- miss_bytes_.fetch_add(size, std::memory_order_relaxed);
- }
/// Mark, publish and fetch the pending ranges at the given indices.
///
@@ -138,6 +113,10 @@ class ReadAheadCache::Impl {
/// re-fetching the same bytes.
void Cache(std::vector<size_t> 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<InputStream> stream_;
CacheConfig config_;
// Ordered by offset (so as to find a matching region by binary search)
@@ -147,18 +126,18 @@ class ReadAheadCache::Impl {
std::vector<std::atomic<bool>> is_cached_;
std::vector<ByteRange> pending_ranges_;
bool is_initialized_ = false;
- // Statistics of the Read() requests issued to the cache, aggregated over
- // all streams sharing this cache.
- std::atomic<uint64_t> read_count_{0};
- std::atomic<uint64_t> read_bytes_{0};
- std::atomic<uint64_t> hits_{0};
- std::atomic<uint64_t> hit_bytes_{0};
- std::atomic<uint64_t> misses_{0};
- std::atomic<uint64_t> miss_bytes_{0};
- // Prefetch IO statistics: how many requests and bytes were actually issued
- // to the underlying stream.
- std::atomic<uint64_t> io_count_{0};
- std::atomic<uint64_t> io_bytes_{0};
+ // 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<FileBlockCache> block_cache_;
+ // The Read() requests issued to the cache and how they were served,
+ // aggregated over all the streams sharing this cache. A read is counted
+ // either as a hit, a block cache hit or a miss.
+ AtomicCounterPair reads_;
+ AtomicCounterPair hits_;
+ AtomicCounterPair misses_;
+ // The prefetch IO actually issued to the underlying stream. The block
cache
+ // counts its own fetches, which CollectMetrics() adds to these.
+ AtomicCounterPair ios_;
};
void ReadAheadCache::Impl::Cache(std::vector<size_t> pending_indices) {
@@ -178,7 +157,7 @@ void ReadAheadCache::Impl::Cache(std::vector<size_t>
pending_indices) {
const ByteRange& range = pending_ranges_[idx];
auto promise = std::make_shared<std::promise<Status>>();
auto future = promise->get_future();
- auto buffer = std::make_shared<Bytes>(range.length,
memory_pool_.get());
+ auto buffer = AllocateBytesKeepingPoolAlive(range.length,
memory_pool_);
fetches.push_back({range, buffer, promise});
new_entries.emplace_back(range, std::move(buffer),
std::move(future));
}
@@ -242,33 +221,50 @@ void ReadAheadCache::Impl::PreBuffer(uint64_t offset) {
}
ReadAheadCache::Impl::Impl(const std::shared_ptr<InputStream>& stream, const
CacheConfig& config,
- const std::shared_ptr<MemoryPool>& memory_pool)
- : stream_(stream), config_(config), memory_pool_(memory_pool) {}
+ uint64_t file_size, const
std::shared_ptr<MemoryPool>& 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<FileBlockCache>(stream, file_size,
config_.GetBlockSize(),
+
config_.GetBlockCacheLimit(), memory_pool);
+ }
+}
ReadAheadCache::Impl::~Impl() {
std::unique_lock<std::shared_mutex> 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();
- read_count_.store(0, std::memory_order_relaxed);
- read_bytes_.store(0, std::memory_order_relaxed);
- hits_.store(0, std::memory_order_relaxed);
- hit_bytes_.store(0, std::memory_order_relaxed);
- misses_.store(0, std::memory_order_relaxed);
- miss_bytes_.store(0, std::memory_order_relaxed);
- io_count_.store(0, std::memory_order_relaxed);
- io_bytes_.store(0, std::memory_order_relaxed);
+ ReleasePrefetchBuffers();
+ reads_.Reset();
+ hits_.Reset();
+ misses_.Reset();
+ ios_.Reset();
+ 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<std::shared_mutex> lock(rw_mutex_);
// Entries are never evicted, so waiting on entries_ covers every
- // dispatched fetch: no async callback can outlive the stream or the
- // memory pool its buffer belongs to.
+ // dispatched fetch: no fetch is still writing into an entry buffer when
the
+ // buffers go away. The buffers keep the memory pool alive themselves, for
+ // the callbacks that a stream destroys later than it resolves them.
for (auto& entry : entries_) {
entry.future.wait();
}
@@ -285,16 +281,22 @@ void
ReadAheadCache::Impl::CollectMetrics(std::shared_ptr<Metrics>* metrics) con
return;
}
auto& m = *metrics;
- m->SetCounter(ReadAheadCacheMetrics::READ_COUNT,
read_count_.load(std::memory_order_relaxed));
- m->SetCounter(ReadAheadCacheMetrics::READ_BYTES,
read_bytes_.load(std::memory_order_relaxed));
- m->SetCounter(ReadAheadCacheMetrics::READ_HITS,
hits_.load(std::memory_order_relaxed));
- m->SetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES,
- hit_bytes_.load(std::memory_order_relaxed));
- 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));
+ m->SetCounter(ReadAheadCacheMetrics::READ_COUNT, reads_.Count());
+ m->SetCounter(ReadAheadCacheMetrics::READ_BYTES, reads_.Bytes());
+ m->SetCounter(ReadAheadCacheMetrics::READ_HITS, hits_.Count());
+ m->SetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES, hits_.Bytes());
+ m->SetCounter(ReadAheadCacheMetrics::READ_MISSES, misses_.Count());
+ m->SetCounter(ReadAheadCacheMetrics::READ_MISS_BYTES, misses_.Bytes());
+ // 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, ios_.Count() +
blocks.fetches);
+ m->SetCounter(ReadAheadCacheMetrics::IO_BYTES, ios_.Bytes() +
blocks.fetch_bytes);
}
void ReadAheadCache::Impl::Warmup() {
@@ -345,12 +347,18 @@ Result<bool> ReadAheadCache::Impl::Read(const ByteRange&
range, char* dest) {
if (range.length == 0) {
return true;
}
- read_count_.fetch_add(1, std::memory_order_relaxed);
- read_bytes_.fetch_add(range.length, std::memory_order_relaxed);
+ reads_.Add(range.length);
PreBuffer(range.offset);
std::vector<RangeCacheEntry> covering = FindCoveringEntries(range);
if (covering.empty()) {
- CountMiss(range.length);
+ // 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;
+ }
+ misses_.Add(range.length);
return false;
}
// Wait OUTSIDE the lock: the futures resolve when the prefetch stream's
@@ -360,7 +368,7 @@ Result<bool> ReadAheadCache::Impl::Read(const ByteRange&
range, char* dest) {
}
// The data copy runs OUTSIDE the lock for the same reason.
CopyRangeFromEntries(covering, range, dest);
- CountHit(range.length);
+ hits_.Add(range.length);
return true;
}
@@ -373,15 +381,14 @@ void ReadAheadCache::Impl::DispatchFetches(const
std::vector<PendingFetch>& fetc
stream_->ReadAsync(
buffer->data(), read_size, read_offset,
[promise, buffer](Status status) mutable {
promise->set_value(status); });
- io_count_.fetch_add(1, std::memory_order_relaxed);
- io_bytes_.fetch_add(fetch.range.length, std::memory_order_relaxed);
+ ios_.Add(fetch.range.length);
}
}
ReadAheadCache::ReadAheadCache(const std::shared_ptr<InputStream>& stream,
- const CacheConfig& config,
+ const CacheConfig& config, uint64_t file_size,
const std::shared_ptr<MemoryPool>& memory_pool)
- : impl_(std::make_unique<Impl>(stream, config, memory_pool)) {}
+ : impl_(std::make_unique<Impl>(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 f039c22e..e44c277d 100644
--- a/src/paimon/common/utils/read_ahead_cache.h
+++ b/src/paimon/common/utils/read_ahead_cache.h
@@ -48,9 +48,31 @@ 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 prefetch IO requests actually issued to the underlying
stream.
+ /// 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` for the
reads
+ /// that complete; a read whose prefetch fetch failed is counted in
read.count
+ /// only, as it is served by neither.
+ ///
+ /// A block hit is a read served out of a block, not a read that avoided
IO:
+ /// the read that finds no block waits for the fetch it dispatches and is
+ /// counted here too. Comparing with block.fetches tells the two apart.
+ 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 IO requests the cache itself issued to the underlying stream:
+ /// the prefetch fetches plus the block fetches. The `io.async.*` metrics
of
+ /// the prefetch reader count those same requests one layer below, but they
+ /// also count the async reads a sub-reader falls back to after this cache
+ /// declined them, so `io.async.requests >= io.count` rather than the two
+ /// agreeing.
static inline const char IO_COUNT[] = "read-ahead-cache.io.count";
- /// Total bytes requested by the prefetch IOs issued to the underlying
stream.
+ /// Total bytes requested by the IOs the cache itself issued to the
+ /// underlying stream.
static inline const char IO_BYTES[] = "read-ahead-cache.io.bytes";
};
@@ -85,11 +107,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 - a reader reads the metadata
+/// of its file 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<InputStream>& stream, const
CacheConfig& config,
- const std::shared_ptr<MemoryPool>& memory_pool);
+ uint64_t file_size, const std::shared_ptr<MemoryPool>&
memory_pool);
~ReadAheadCache();
/// Initialize the cache with given byte ranges to be cached.
@@ -104,9 +139,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<bool> Read(const ByteRange& range, char* dest);
@@ -115,11 +153,11 @@ class PAIMON_EXPORT ReadAheadCache {
/// when the first Read() arrives, racing the caller's own miss fetch.
void Warmup();
- /// Collect hit/miss counters of Read() calls and the prefetch IO
- /// counters into the given metrics as counters named after
+ /// Collect the counters of the Read() calls, of the block cache and of the
+ /// IOs into the given metrics as counters named after
/// `ReadAheadCacheMetrics`. Only reads issued through Read() are counted
- /// as hits/misses; prefetch fetches dispatched by the cache itself are
- /// counted in the fetch counters instead.
+ /// as hits, block hits or misses; the fetches the cache dispatches itself
+ /// are counted in the io counters instead.
/// @param metrics The metrics to write the counters into. A null
/// pointer or a null shared pointer is a no-op.
void CollectMetrics(std::shared_ptr<Metrics>* metrics) const;
@@ -129,6 +167,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 +177,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 e7900b7b..60fe8709 100644
--- a/src/paimon/common/utils/read_ahead_cache_test.cpp
+++ b/src/paimon/common/utils/read_ahead_cache_test.cpp
@@ -21,7 +21,7 @@
#include <chrono>
#include <fstream>
-#include <mutex>
+#include <memory>
#include <thread>
#include <vector>
@@ -31,40 +31,62 @@
#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<paimon::ReadAheadCache> cache;
- std::shared_ptr<paimon::MemoryPool> 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<paimon::ByteRange> ranges) {
- auto dir = UniqueTestDirectory::Create();
- EXPECT_TRUE(dir);
- std::string path = dir->Str() + "/" + filename;
+// 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();
- 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<FileSystem> fs,
FileSystemFactory::Get("local", path, {}));
+ EXPECT_OK_AND_ASSIGN(std::unique_ptr<InputStream> in, fs->Open(path));
+ return std::move(in);
+}
- auto pool = GetDefaultPool();
- auto cache = std::make_shared<ReadAheadCache>(std::move(in), config, pool);
+// 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->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 +96,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 +105,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<InputStream> inner) :
inner_(std::move(inner)) {}
-
- Status Close() override {
- return inner_->Close();
- }
- Status Seek(int64_t offset, SeekOrigin origin) override {
- return inner_->Seek(offset, origin);
- }
- Result<int64_t> GetPos() const override {
- return inner_->GetPos();
- }
- Result<int64_t> Read(char* buffer, int64_t size) override {
- return inner_->Read(buffer, size);
- }
- Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override {
- return inner_->Read(buffer, size, offset);
- }
- void ReadAsync(char* buffer, int64_t size, int64_t offset,
- std::function<void(Status)>&& callback) override {
- std::lock_guard<std::mutex> lock(mutex_);
- async_read_count_++;
- pending_.push_back({buffer, size, offset, std::move(callback)});
- }
- Result<std::string> GetUri() const override {
- return inner_->GetUri();
- }
- Result<int64_t> Length() const override {
- return inner_->Length();
- }
-
- int AsyncReadCount() {
- std::lock_guard<std::mutex> lock(mutex_);
- return async_read_count_;
- }
-
- /// Complete all held fetches against the underlying stream.
- void ReleaseAll() {
- std::vector<PendingRead> taken;
- {
- std::lock_guard<std::mutex> lock(mutex_);
- taken = std::move(pending_);
- pending_.clear();
- }
- for (auto& read : taken) {
- Result<int64_t> res = inner_->Read(read.buffer, read.size,
read.offset);
- read.callback(res.ok() ? Status::OK() : res.status());
- }
- }
-
- private:
- struct PendingRead {
- char* buffer;
- int64_t size;
- int64_t offset;
- std::function<void(Status)> callback;
- };
-
- std::shared_ptr<InputStream> inner_;
- std::mutex mutex_;
- std::vector<PendingRead> pending_;
- int async_read_count_ = 0;
-};
-
TEST(TestReadAheadCache, TestBasics) {
- CacheConfig config(/*range_size_limit=*/10,
- /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024
* 1024);
+ 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<ReadAheadCache> 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 +139,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<ReadAheadCache> cache_ptr =
+ CreateTestFileAndCache("data_file", content, config, {{0, 25}});
+ ReadAheadCache& cache = *cache_ptr;
// Spans all three entries.
AssertReadEquals({5, 20}, "fghijklmnopqrstuvwxy", &cache);
@@ -217,11 +174,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<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);
@@ -230,11 +188,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<ReadAheadCache> cache_ptr =
+ CreateTestFileAndCache("data_file", content, config, {{0, 5}, {8, 5},
{16, 5}});
+ ReadAheadCache& cache = *cache_ptr;
AssertReadEquals({0, 5}, "abcde", &cache);
@@ -246,11 +205,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<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.
@@ -286,11 +247,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<ReadAheadCache> cache_ptr =
+ CreateTestFileAndCache("data_file", content, config, {{0, 5}});
+ ReadAheadCache& cache = *cache_ptr;
AssertReadEquals({0, 5}, "abcde", &cache);
@@ -341,28 +304,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<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(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<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(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,56 +336,50 @@ 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<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", 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<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(/*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);
- std::string path = dir->Str() + "/data_file";
- std::ofstream file(path, std::ios::binary);
- ASSERT_TRUE(file.is_open());
- file.write(content.data(), content.size());
- ASSERT_FALSE(file.fail());
- file.close();
- ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", path, {}));
- ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs->Open(path));
- auto gated = std::make_shared<GatedAsyncInputStream>(std::move(in));
+ std::unique_ptr<UniqueTestDirectory> dir;
+ auto gated = std::make_shared<GatedAsyncInputStream>(OpenTestFile(&dir,
"data_file", content));
- ReadAheadCache cache(gated, config, GetDefaultPool());
+ ReadAheadCache cache(gated, config, /*file_size=*/0, TestPool());
ASSERT_OK(cache.Init({{0, 5}}));
cache.Warmup();
@@ -453,14 +412,48 @@ TEST(TestReadAheadCache,
TestInFlightEntryServesRacingReader) {
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.Init({{0, 5}}));
+ 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(/*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<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(); });
@@ -483,11 +476,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<ReadAheadCache> 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 +492,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<ReadAheadCache> cache_ptr =
+ CreateTestFileAndCache("data_file", content, config, {{0, 5}});
+ ReadAheadCache& cache = *cache_ptr;
AssertReadEquals({0, 5}, "abcde", &cache);
@@ -517,25 +512,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<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(/*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<ReadAheadCache> cache_ptr =
+ CreateTestFileAndCache("data_file", content, config, {{0, 5}});
+ cache_ptr->CollectMetrics(/*metrics=*/nullptr);
std::shared_ptr<Metrics> 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> 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 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<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
diff --git a/src/paimon/core/operation/read_context_test.cpp
b/src/paimon/core/operation/read_context_test.cpp
index 1174568f..3d362352 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<MemoryPool> memory_pool = GetDefaultPool();
std::shared_ptr<Executor> 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 c52ceba9..bed06bb7 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 <algorithm>
#include <atomic>
#include <functional>
+#include <future>
#include <iostream>
#include <limits>
#include <memory>
@@ -1794,14 +1795,22 @@ TEST_F(ParquetFileBatchReaderTest,
TestPreBufferRangeFeedsReadAheadCache) {
WriteArray(file_path_, src_array, arrow_schema, /*write_batch_size=*/10,
/*enable_dictionary=*/true, /*max_row_group_length=*/10);
+ // A pool of its own, 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. Declared first, so that it outlives the cache and the reader.
+ std::shared_ptr<MemoryPool> pool(GetMemoryPool());
+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> cache_stream,
fs_->Open(file_path_));
- auto cache = std::make_shared<ReadAheadCache>(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<ReadAheadCache>(cache_stream, CacheConfig(),
/*file_size=*/0, pool);
ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> reader_stream,
fs_->Open(file_path_));
auto cache_input_stream =
std::make_shared<CacheInputStream>(std::move(reader_stream), cache);
std::map<std::string, std::string> options;
ParquetReaderBuilder builder(options, batch_size_);
- builder.WithMemoryPool(GetDefaultPool());
+ builder.WithMemoryPool(pool);
ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileBatchReader> base_reader,
builder.Build(cache_input_stream));
auto parquet_batch_reader =
dynamic_cast<ParquetFileBatchReader*>(base_reader.get());
@@ -2154,4 +2163,133 @@ 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<InputStream> input, int64_t
tail_begin,
+ std::shared_ptr<std::atomic<int32_t>>
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<int64_t> GetPos() const override {
+ return input_->GetPos();
+ }
+ Result<int64_t> Read(char* buffer, int64_t size) override {
+ return input_->Read(buffer, size);
+ }
+ Result<int64_t> 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<void(Status)>&& callback) override {
+ CountIfInTail(offset);
+ return input_->ReadAsync(buffer, size, offset, std::move(callback));
+ }
+ Status Close() override {
+ return input_->Close();
+ }
+ Result<std::string> GetUri() const override {
+ return input_->GetUri();
+ }
+ Result<int64_t> 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<InputStream> input_;
+ int64_t tail_begin_;
+ std::shared_ptr<std::atomic<int32_t>> 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<InputStream> 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<int64_t>(CacheConfig().GetBlockSize());
+ ASSERT_GT(tail_begin, 0);
+
+ // A pool of its own, 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. Declared first, so that it outlives the cache and the
readers.
+ std::shared_ptr<MemoryPool> pool(GetMemoryPool());
+
+ auto tail_read_count = std::make_shared<std::atomic<int32_t>>(0);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> cache_stream,
fs_->Open(file_path_));
+ auto counting_cache_stream = std::make_shared<TailReadCountingInputStream>(
+ std::move(cache_stream), tail_begin, tail_read_count);
+ auto cache = std::make_shared<ReadAheadCache>(counting_cache_stream,
CacheConfig(),
+
static_cast<uint64_t>(file_length), pool);
+
+ std::map<std::string, std::string> options;
+ options[PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] = "true";
+ ParquetReaderBuilder builder(options, batch_size_);
+ builder.WithMemoryPool(pool);
+
+ constexpr int32_t kReaderCount = 3;
+ std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> futures;
+ for (int32_t i = 0; i < kReaderCount; ++i) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> reader_stream,
fs_->Open(file_path_));
+ auto counting =
std::make_unique<TailReadCountingInputStream>(std::move(reader_stream),
+
tail_begin, tail_read_count);
+ std::shared_ptr<InputStream> cache_input_stream =
+ std::make_shared<CacheInputStream>(std::move(counting), cache);
+ futures.push_back(std::async(std::launch::async, [&builder,
cache_input_stream]() {
+ return builder.Build(cache_input_stream);
+ }));
+ }
+ std::vector<std::unique_ptr<FileBatchReader>> readers;
+ for (auto& future : futures) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileBatchReader> 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> predicate = PredicateBuilder::GreaterOrEqual(
+ /*field_index=*/0, /*field_name=*/"f0", FieldType::INT,
Literal(59000));
+ for (auto& reader : readers) {
+ auto parquet_batch_reader =
dynamic_cast<ParquetFileBatchReader*>(reader.get());
+ ASSERT_TRUE(parquet_batch_reader);
+ std::unique_ptr<ArrowSchema> c_schema =
std::make_unique<ArrowSchema>();
+ 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> metrics = std::make_shared<MetricsImpl>();
+ 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 00000000..3115a3e8
--- /dev/null
+++ b/src/paimon/testing/utils/gated_async_input_stream.h
@@ -0,0 +1,139 @@
+/*
+ * 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 <cstdint>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <utility>
+#include <vector>
+
+#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<InputStream> inner) :
inner_(std::move(inner)) {}
+
+ Status Close() override {
+ return inner_->Close();
+ }
+ Status Seek(int64_t offset, SeekOrigin origin) override {
+ return inner_->Seek(offset, origin);
+ }
+ Result<int64_t> GetPos() const override {
+ return inner_->GetPos();
+ }
+ Result<int64_t> Read(char* buffer, int64_t size) override {
+ return inner_->Read(buffer, size);
+ }
+ Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override {
+ return inner_->Read(buffer, size, offset);
+ }
+ void ReadAsync(char* buffer, int64_t size, int64_t offset,
+ std::function<void(Status)>&& callback) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ async_read_count_++;
+ pending_.push_back({buffer, size, offset, std::move(callback)});
+ }
+ Result<std::string> GetUri() const override {
+ return inner_->GetUri();
+ }
+ Result<int64_t> Length() const override {
+ return inner_->Length();
+ }
+
+ int32_t AsyncReadCount() const {
+ std::lock_guard<std::mutex> lock(mutex_);
+ return async_read_count_;
+ }
+
+ /// Complete all held fetches against the underlying stream.
+ void ReleaseAll() {
+ for (auto& read : TakePending()) {
+ Result<int64_t> res = inner_->Read(read.buffer, read.size,
read.offset);
+ read.callback(res.ok() ? Status::OK() : res.status());
+ }
+ }
+
+ /// Complete all held fetches against the underlying stream but keep their
+ /// callbacks alive, the way an object store stream destroys the callback
of a
+ /// read later than it resolves it: whatever the callback captured stays
alive
+ /// until DropCompletedCallbacks() is called.
+ void ReleaseAllKeepingCallbacks() {
+ std::vector<PendingRead> taken = TakePending();
+ for (auto& read : taken) {
+ Result<int64_t> res = inner_->Read(read.buffer, read.size,
read.offset);
+ read.callback(res.ok() ? Status::OK() : res.status());
+ }
+ std::lock_guard<std::mutex> lock(mutex_);
+ for (auto& read : taken) {
+ completed_.push_back(std::move(read));
+ }
+ }
+
+ /// Destroy the callbacks kept alive by ReleaseAllKeepingCallbacks().
+ void DropCompletedCallbacks() {
+ std::vector<PendingRead> taken;
+ {
+ std::lock_guard<std::mutex> lock(mutex_);
+ taken = std::move(completed_);
+ completed_.clear();
+ }
+ }
+
+ /// 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) {
+ for (auto& read : TakePending()) {
+ read.callback(status);
+ }
+ }
+
+ private:
+ struct PendingRead {
+ char* buffer;
+ int64_t size;
+ int64_t offset;
+ std::function<void(Status)> callback;
+ };
+
+ std::vector<PendingRead> TakePending() {
+ std::lock_guard<std::mutex> lock(mutex_);
+ std::vector<PendingRead> taken = std::move(pending_);
+ pending_.clear();
+ return taken;
+ }
+
+ std::shared_ptr<InputStream> inner_;
+ mutable std::mutex mutex_;
+ std::vector<PendingRead> pending_;
+ // The fetches completed by ReleaseAllKeepingCallbacks(), held to keep
their
+ // callbacks alive.
+ std::vector<PendingRead> completed_;
+ int32_t async_read_count_ = 0;
+};
+
+} // namespace paimon::test
