feat(core): add a block cache for the reads the prefetched ranges do not cover

include/paimon/utils/prefetch_cache_config.h

@@ -34,10 +34,8 @@ namespace paimon {
 /// ReadAheadCache to balance memory usage, I/O efficiency, and latency hiding.
 class PAIMON_EXPORT CacheConfig {
  public:
-    CacheConfig();
-    CacheConfig(uint64_t range_size_limit, uint64_t hole_size_limit, uint64_t pre_buffer_limit);
-
     /// Returns the maximum allowed size (in bytes) for a single cached range.
+    /// Defaults to 32 MiB.
     uint64_t GetRangeSizeLimit() const {
         return range_size_limit_;
     }
@@ -47,7 +45,8 @@ class PAIMON_EXPORT CacheConfig {
         range_size_limit_ = range_size_limit;
     }
 
-    /// Returns the maximum gap size (in bytes) considered mergeable between adjacent ranges.
+    /// Returns the maximum gap size (in bytes) considered mergeable between
+    /// adjacent ranges. Defaults to 8 KiB.
     uint64_t GetHoleSizeLimit() const {
         return hole_size_limit_;
     }
@@ -57,7 +56,8 @@ class PAIMON_EXPORT CacheConfig {
         hole_size_limit_ = hole_size_limit;
     }
 
-    /// Returns the maximum size to pre-buffer ahead of the current read position.
+    /// Returns the maximum size to pre-buffer ahead of the current read
+    /// position. Defaults to 256 MiB.
     uint64_t GetPreBufferLimit() const {
         return pre_buffer_limit_;
     }
@@ -67,10 +67,50 @@ class PAIMON_EXPORT CacheConfig {
         pre_buffer_limit_ = pre_buffer_limit;
     }
 
+    /// Returns the granularity (in bytes) of the block cache entries serving the
+    /// small reads that the prefetched ranges do not cover. Defaults to 64 KiB.
+    uint64_t GetBlockSize() const {
+        return block_size_;
+    }
+
+    /// Sets the granularity (in bytes) of the block cache entries.
+    void SetBlockSize(uint64_t block_size) {
+        block_size_ = block_size;
+    }
+
+    /// Returns the maximum total size (in bytes) of the block cache entries of
+    /// one file. Zero disables the block cache. Defaults to 1 MiB.
+    uint64_t GetBlockCacheLimit() const {
+        return block_cache_limit_;
+    }
+
+    /// Sets the maximum total size (in bytes) of the block cache entries of one
+    /// file. Zero disables the block cache.
+    void SetBlockCacheLimit(uint64_t block_cache_limit) {
+        block_cache_limit_ = block_cache_limit;
+    }
+
  private:
-    uint64_t range_size_limit_;
-    uint64_t hole_size_limit_;
-    uint64_t pre_buffer_limit_;
+    // The defaults are aligned with the reader's request granularity and with
+    // realistic data file sizes:
+    // - range_size_limit matches the parquet reader's 32 MiB request blocks
+    //   (Arrow ReadRangeCache's own range limit); a smaller limit cuts entries
+    //   below the request size, so a request can never be served from one piece.
+    // - pre_buffer_limit must exceed the LARGEST single read a reader issues
+    //   (coalesced column-chunk reads of ~128 MiB were observed): fetches are
+    //   only dispatched up to this window, so a request reaching past it can
+    //   never be served and falls back to a second fetch of the same bytes.
+    uint64_t range_size_limit_ = 32 * 1024 * 1024;
+    uint64_t hole_size_limit_ = 8 * 1024;
+    uint64_t pre_buffer_limit_ = 256 * 1024 * 1024;
+    // Blocks are aligned to the END of the file, so a block never reaches past
+    // EOF. 64 KiB matches the footer read of the parquet reader (arrow's
+    // kDefaultFooterReadSize), which then covers exactly one block instead of
+    // straddling two.
+    uint64_t block_size_ = 64 * 1024;
+    // One block is enough for the metadata tail of a file; the limit only
+    // bounds the pathological case, as blocks are never evicted.
+    uint64_t block_cache_limit_ = 1024 * 1024;
 };
 
 }  // namespace paimon

src/paimon/CMakeLists.txt

@@ -179,6 +179,7 @@ set(PAIMON_COMMON_SRCS
     common/data/shredding/shredding_file_reader.cpp
     common/utils/delta_varint_compressor.cpp
     common/utils/fields_comparator.cpp
+    common/utils/file_block_cache.cpp
     common/utils/path_util.cpp
     common/utils/range.cpp
     common/utils/read_ahead_cache.cpp
@@ -667,6 +668,7 @@ if(PAIMON_BUILD_TESTS)
                     common/utils/roaring_bitmap64_test.cpp
                     common/utils/range_helper_test.cpp
                     common/utils/read_ahead_cache_test.cpp
+                    common/utils/file_block_cache_test.cpp
                     common/io/cache/lru_cache_test.cpp
                     common/io/cache/cache_manager_test.cpp
                     common/utils/byte_range_combiner_test.cpp

src/paimon/common/io/cache_input_stream_test.cpp

@@ -60,9 +60,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 +209,12 @@ TEST_F(CacheInputStreamTest, TestReadAsyncCacheReadError) {
         ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", file_path_, {}));
         ASSERT_OK_AND_ASSIGN(auto cache_stream, fs->Open(file_path_));
         ASSERT_OK_AND_ASSIGN(auto underlying, fs->Open(file_path_));
-        CacheConfig config(/*range_size_limit=*/1024,
-                           /*hole_size_limit=*/0, /*pre_buffer_limit=*/1024 * 1024);
-        auto cache = std::make_shared<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)

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++) {

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_,

src/paimon/common/utils/file_block_cache.cpp

@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/common/utils/file_block_cache.h"
+
+#include <cstring>
+
+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 or the memory pool.
+    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;
+            block->buffer = std::make_shared<Bytes>(block_range.length, memory_pool_.get());
+            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_.fetch_add(1, std::memory_order_relaxed);
+    hit_bytes_.fetch_add(range.length, std::memory_order_relaxed);
+    return true;
+}
+
+void FileBlockCache::Release() {
+    std::lock_guard<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_.store(0, std::memory_order_relaxed);
+    hit_bytes_.store(0, std::memory_order_relaxed);
+    fetches_.store(0, std::memory_order_relaxed);
+    fetch_bytes_.store(0, std::memory_order_relaxed);
+}
+
+FileBlockCache::Counters FileBlockCache::GetCounters() const {
+    Counters counters;
+    counters.hits = hits_.load(std::memory_order_relaxed);
+    counters.hit_bytes = hit_bytes_.load(std::memory_order_relaxed);
+    counters.fetches = fetches_.load(std::memory_order_relaxed);
+    counters.fetch_bytes = fetch_bytes_.load(std::memory_order_relaxed);
+    return counters;
+}
+
+bool FileBlockCache::CanServe(const ByteRange& range) const {
+    if (capacity_ == 0 || block_size_ == 0 || file_size_ == 0) {
+        return false;
+    }
+    if (range.length == 0 || range.length > block_size_) {
+        return false;
+    }
+    // A read reaching past EOF is left to the caller: serving it would mean
+    // short-reading into the block buffer.
+    if (range.offset >= file_size_ || range.length > file_size_ - range.offset) {
+        return false;
+    }
+    // A read straddling two blocks would need both of them to be present; it is
+    // left to the caller instead, which keeps one block per served read.
+    return IndexOf(range.offset) == IndexOf(range.offset + range.length - 1);
+}
+
+uint64_t FileBlockCache::IndexOf(uint64_t offset) const {
+    // Counted from the end of the file, so that block 0 is the last block.
+    return (file_size_ - 1 - offset) / block_size_;
+}
+
+ByteRange FileBlockCache::RangeOf(uint64_t index) const {
+    const uint64_t end = file_size_ - index * block_size_;
+    const uint64_t offset = end > block_size_ ? end - block_size_ : 0;
+    return {offset, end - offset};
+}
+
+void FileBlockCache::Fetch(const std::shared_ptr<Block>& block) {
+    fetches_.fetch_add(1, std::memory_order_relaxed);
+    fetch_bytes_.fetch_add(block->range.length, std::memory_order_relaxed);
+    auto promise = block->promise;
+    auto buffer = block->buffer;
+    // The buffer and the promise are captured, so the async read keeps its
+    // destination and the future it resolves alive.
+    stream_->ReadAsync(buffer->data(), static_cast<int64_t>(buffer->size()),
+                       static_cast<int64_t>(block->range.offset),
+                       [promise, buffer](Status status) { promise->set_value(status); });
+}
+
+}  // namespace paimon