Skip to content

Commit a2859e7

Browse files
committed
fix
1 parent 5112cdc commit a2859e7

5 files changed

Lines changed: 59 additions & 33 deletions

File tree

‎include/paimon/utils/prefetch_cache_config.h‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -104,9 +104,10 @@ class PAIMON_EXPORT CacheConfig {
104104
uint64_t hole_size_limit_ = 8 * 1024;
105105
uint64_t pre_buffer_limit_ = 256 * 1024 * 1024;
106106
// Blocks are aligned to the END of the file, so a block never reaches past
107-
// EOF. 64 KiB matches the footer read of the parquet reader (arrow's
108-
// kDefaultFooterReadSize), which then covers exactly one block instead of
109-
// straddling two.
107+
// EOF. 64 KiB is the granularity the reads no prefetched range covers are
108+
// shared at: small enough that a metadata read at the tail of a file is
109+
// served by one block instead of straddling two, large enough that a block
110+
// fetch does not pull in much more than the reads ask for.
110111
uint64_t block_size_ = 64 * 1024;
111112
// One block is enough for the metadata tail of a file; the limit only
112113
// bounds the pathological case, as blocks are never evicted.

‎src/paimon/common/io/cache_input_stream_test.cpp‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,10 @@ namespace paimon::test {
4040
class CacheInputStreamTest : public ::testing::Test {
4141
public:
4242
void SetUp() override {
43-
pool_ = GetDefaultPool();
43+
// A pool of its own, so that a cache buffer outliving the pool it was
44+
// allocated from shows up instead of being covered by the global pool,
45+
// which never goes away.
46+
pool_ = std::shared_ptr<MemoryPool>(GetMemoryPool());
4447
test_dir_ = UniqueTestDirectory::Create();
4548
ASSERT_TRUE(test_dir_);
4649
content_ = "abcdefghijklmnopqrstuvwxyz0123456789";

‎src/paimon/common/utils/file_block_cache.h‎

Lines changed: 19 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -35,18 +35,27 @@
3535

3636
namespace paimon {
3737

38-
/// A cache of fixed-size blocks of one file, serving the reads that no
39-
/// prefetched range covers: a parquet reader reads the footer and the page index
40-
/// before any range can be registered, and every reader of the file reads the
41-
/// same bytes.
38+
/// A cache of fixed-size blocks of one file, the fallback below the prefetched
39+
/// ranges: a read that no registered range covers is served here at block
40+
/// granularity instead of being left to the caller, which would read it from the
41+
/// underlying stream uncached and pay for it again on the next reader. It is a
42+
/// general mechanism over the whole file - any reader, any offset, any round of
43+
/// registered ranges - not a cache of one particular kind of read.
4244
///
4345
/// Blocks are aligned to the END of the file: block 0 is
44-
/// [file_size - block_size, file_size). The metadata of a parquet/orc file lives
45-
/// in its tail and arrow reads exactly the last 64 KiB as the footer, so an
46-
/// end-aligned block matches that read instead of straddling two blocks. It also
47-
/// keeps every block inside the file, as long as the given file size is the size
48-
/// of the file: a block fetch failing because the file is shorter than that only
49-
/// costs the caching of that block, see Read().
46+
/// [file_size - block_size, file_size). The reads served here concentrate at the
47+
/// tail of a file: a reader reads the metadata of its file before it knows which
48+
/// ranges to register, and every reader of the file reads the same bytes. End
49+
/// alignment gives that region whole blocks - a read of the last block_size bytes
50+
/// of a file is one block whatever the file size - and leaves the only partial
51+
/// block at the head of the file. It also keeps every block inside the file, as
52+
/// long as the given file size is the size of the file: a block fetch failing
53+
/// because the file is shorter than that only costs the caching of that block,
54+
/// see Read().
55+
///
56+
/// One block serves one read: a read longer than a block, or straddling two, is
57+
/// declined and left to the caller instead of being fetched in pieces, so a
58+
/// served read stays one block, one fetch and one copy.
5059
///
5160
/// A block is published before its fetch is dispatched, so concurrent readers of
5261
/// the same block wait for that one fetch instead of issuing their own.

‎src/paimon/common/utils/read_ahead_cache.h‎

Lines changed: 16 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -64,12 +64,15 @@ class PAIMON_EXPORT ReadAheadCacheMetrics {
6464
/// the block granularity added on top of the requested ones.
6565
static inline const char BLOCK_FETCHES[] = "read-ahead-cache.block.fetches";
6666
static inline const char BLOCK_FETCH_BYTES[] = "read-ahead-cache.block.fetch-bytes";
67-
/// Number of prefetch IO requests actually issued to the underlying stream.
68-
/// These are the same requests that the `io.async.*` metrics of the prefetch
69-
/// reader observe one layer below, counted here per cache rather than per
70-
/// stream, so the two are expected to agree instead of adding up.
67+
/// Number of IO requests the cache itself issued to the underlying stream:
68+
/// the prefetch fetches plus the block fetches. The `io.async.*` metrics of
69+
/// the prefetch reader count those same requests one layer below, but they
70+
/// also count the async reads a sub-reader falls back to after this cache
71+
/// declined them, so `io.async.requests >= io.count` rather than the two
72+
/// agreeing.
7173
static inline const char IO_COUNT[] = "read-ahead-cache.io.count";
72-
/// Total bytes requested by the prefetch IOs issued to the underlying stream.
74+
/// Total bytes requested by the IOs the cache itself issued to the
75+
/// underlying stream.
7376
static inline const char IO_BYTES[] = "read-ahead-cache.io.bytes";
7477
};
7578

@@ -105,10 +108,10 @@ struct PAIMON_EXPORT ByteRange {
105108
/// ReleaseBuffers() or Reset(). It is meant to hold the prefetched ranges of
106109
/// a single data file, whose size is bounded by the reader's scan scope.
107110
///
108-
/// Reads that the prefetched ranges do not cover (the footer and the page index
109-
/// of a parquet file are read before any range is registered) are served by a
110-
/// FileBlockCache instead of being left to the caller. That block cache is owned
111-
/// by this one and shares its lifetime: it is configured from `block_size` and
111+
/// Reads that the prefetched ranges do not cover - a reader reads the metadata
112+
/// of its file before any range is registered - are served by a FileBlockCache
113+
/// instead of being left to the caller. That block cache is owned by this one
114+
/// and shares its lifetime: it is configured from `block_size` and
112115
/// `block_cache_limit`, it survives Reset() - the blocks belong to the file
113116
/// rather than to a registration round - and it is released by ReleaseBuffers().
114117
class PAIMON_EXPORT ReadAheadCache {
@@ -150,11 +153,11 @@ class PAIMON_EXPORT ReadAheadCache {
150153
/// when the first Read() arrives, racing the caller's own miss fetch.
151154
void Warmup();
152155

153-
/// Collect hit/miss counters of Read() calls and the prefetch IO
154-
/// counters into the given metrics as counters named after
156+
/// Collect the counters of the Read() calls, of the block cache and of the
157+
/// IOs into the given metrics as counters named after
155158
/// `ReadAheadCacheMetrics`. Only reads issued through Read() are counted
156-
/// as hits/misses; prefetch fetches dispatched by the cache itself are
157-
/// counted in the fetch counters instead.
159+
/// as hits, block hits or misses; the fetches the cache dispatches itself
160+
/// are counted in the io counters instead.
158161
/// @param metrics The metrics to write the counters into. A null
159162
/// pointer or a null shared pointer is a no-op.
160163
void CollectMetrics(std::shared_ptr<Metrics>* metrics) const;

‎src/paimon/format/parquet/parquet_file_batch_reader_test.cpp‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1679,17 +1679,22 @@ TEST_F(ParquetFileBatchReaderTest, TestPreBufferRangeFeedsReadAheadCache) {
16791679
WriteArray(file_path_, src_array, arrow_schema, /*write_batch_size=*/10,
16801680
/*enable_dictionary=*/true, /*max_row_group_length=*/10);
16811681

1682+
// A pool of its own, so that a buffer outliving the pool it was allocated
1683+
// from shows up instead of being covered by the global pool, which never
1684+
// goes away. Declared first, so that it outlives the cache and the reader.
1685+
std::shared_ptr<MemoryPool> pool(GetMemoryPool());
1686+
16821687
ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> cache_stream, fs_->Open(file_path_));
16831688
// The file size is left unknown so the block cache stays off: this test is
16841689
// about the pre-buffered ranges feeding the cache.
1685-
auto cache = std::make_shared<ReadAheadCache>(cache_stream, CacheConfig(), /*file_size=*/0,
1686-
GetDefaultPool());
1690+
auto cache =
1691+
std::make_shared<ReadAheadCache>(cache_stream, CacheConfig(), /*file_size=*/0, pool);
16871692
ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> reader_stream, fs_->Open(file_path_));
16881693
auto cache_input_stream = std::make_shared<CacheInputStream>(std::move(reader_stream), cache);
16891694

16901695
std::map<std::string, std::string> options;
16911696
ParquetReaderBuilder builder(options, batch_size_);
1692-
builder.WithMemoryPool(GetDefaultPool());
1697+
builder.WithMemoryPool(pool);
16931698
ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileBatchReader> base_reader,
16941699
builder.Build(cache_input_stream));
16951700
auto parquet_batch_reader = dynamic_cast<ParquetFileBatchReader*>(base_reader.get());
@@ -2111,17 +2116,22 @@ TEST_F(ParquetFileBatchReaderTest, TestBlockCacheSharesTailReadsAcrossReaders) {
21112116
const int64_t tail_begin = file_length - static_cast<int64_t>(CacheConfig().GetBlockSize());
21122117
ASSERT_GT(tail_begin, 0);
21132118

2119+
// A pool of its own, so that a buffer outliving the pool it was allocated
2120+
// from shows up instead of being covered by the global pool, which never
2121+
// goes away. Declared first, so that it outlives the cache and the readers.
2122+
std::shared_ptr<MemoryPool> pool(GetMemoryPool());
2123+
21142124
auto tail_read_count = std::make_shared<std::atomic<int32_t>>(0);
21152125
ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> cache_stream, fs_->Open(file_path_));
21162126
auto counting_cache_stream = std::make_shared<TailReadCountingInputStream>(
21172127
std::move(cache_stream), tail_begin, tail_read_count);
2118-
auto cache = std::make_shared<ReadAheadCache>(
2119-
counting_cache_stream, CacheConfig(), static_cast<uint64_t>(file_length), GetDefaultPool());
2128+
auto cache = std::make_shared<ReadAheadCache>(counting_cache_stream, CacheConfig(),
2129+
static_cast<uint64_t>(file_length), pool);
21202130

21212131
std::map<std::string, std::string> options;
21222132
options[PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] = "true";
21232133
ParquetReaderBuilder builder(options, batch_size_);
2124-
builder.WithMemoryPool(GetDefaultPool());
2134+
builder.WithMemoryPool(pool);
21252135

21262136
constexpr int32_t kReaderCount = 3;
21272137
std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> futures;

0 commit comments

Comments
 (0)