fix

Author: lucasfang

include/paimon/reader/prefetch_file_batch_reader.h

@@ -137,18 +137,19 @@ class PAIMON_EXPORT PrefetchFileBatchReader : public FileBatchReader {
         return std::vector<std::pair<uint64_t, uint64_t>>{};
     }
 
-    /// Sink a reader reports byte ranges to when they only become known after reading has
+    /// Callback a reader reports byte ranges through when they only become known after reading has
     /// started, so the prefetch layer can register them with its shared read-ahead cache.
     ///
     /// PreBufferRange() covers what is known up front; a reader whose ranges depend on data it
     /// has already read - the late-materialization payload pass only knows which pages hold the
-    /// matched rows once the probe pass has run - reports them here instead.
-    using PreBufferSink = std::function<void(std::vector<std::pair<uint64_t, uint64_t>>&&)>;
+    /// matched rows once the probe pass has run - reports them through this callback instead.
+    using PreBufferRangeCallback =
+        std::function<void(std::vector<std::pair<uint64_t, uint64_t>>&&)>;
 
-    /// Installs the sink above, or clears it when `sink` is empty. By default a reader has no
-    /// late byte ranges to report and ignores the sink.
-    /// @param sink The sink to report late byte ranges to.
-    virtual void SetPreBufferSink(PreBufferSink sink) {}
+    /// Installs the callback above, or clears it when `callback` is empty. By default a reader has
+    /// no late byte ranges to report and ignores the callback.
+    /// @param callback The callback to report late byte ranges through.
+    virtual void SetPreBufferRangeCallback(PreBufferRangeCallback callback) {}
 };
 
 }  // namespace paimon

src/paimon/common/reader/late_materializing_file_batch_reader.cpp

@@ -268,7 +268,7 @@ Status LateMaterializingFileBatchReader::ReportPayloadPreBufferRanges() {
     // reader's target row groups down to the pages holding the matched rows. Reporting them
     // here lets the shared read-ahead cache fetch them while this pass is still assembling
     // its first batch, instead of every payload read waiting for its own IO.
-    if (!pre_buffer_sink_ || prefetch_inner_ == nullptr) {
+    if (!pre_buffer_range_callback_ || prefetch_inner_ == nullptr) {
         return Status::OK();
     }
     // The error is propagated rather than swallowed: the ranges come from the file metadata, so a
@@ -278,7 +278,7 @@ Status LateMaterializingFileBatchReader::ReportPayloadPreBufferRanges() {
     if (ranges.empty()) {
         return Status::OK();
     }
-    pre_buffer_sink_(std::move(ranges));
+    pre_buffer_range_callback_(std::move(ranges));
     return Status::OK();
 }
 

src/paimon/common/reader/late_materializing_file_batch_reader.h

@@ -95,18 +95,18 @@ class LateMaterializingFileBatchReader : public PrefetchFileBatchReader {
     Result<std::vector<std::pair<uint64_t, uint64_t>>> PreBufferRange() override {
         // Deliberately only the probe ranges: this is asked once before any read, when the payload
         // pass still has no idea which pages hold the matched rows. The payload ranges are
-        // reported through pre_buffer_sink_ once the probe pass has run, see
+        // reported through pre_buffer_range_callback_ once the probe pass has run, see
         // ReportPayloadPreBufferRanges().
         if (prefetch_inner_ == nullptr) {
             return std::vector<std::pair<uint64_t, uint64_t>>{};
         }
         return prefetch_inner_->PreBufferRange();
     }
 
-    void SetPreBufferSink(PreBufferSink sink) override {
+    void SetPreBufferRangeCallback(PreBufferRangeCallback callback) override {
         // Not forwarded to prefetch_inner_: a format reader has no late ranges of its own, and
         // forwarding would report the same ranges twice.
-        pre_buffer_sink_ = std::move(sink);
+        pre_buffer_range_callback_ = std::move(callback);
     }
 
  private:
@@ -149,9 +149,9 @@ class LateMaterializingFileBatchReader : public PrefetchFileBatchReader {
                               const std::shared_ptr<Predicate>& predicate,
                               const std::optional<RoaringBitmap32>& selection);
 
-    /// Report the byte ranges the payload pass is about to read to pre_buffer_sink_, so that the
-    /// prefetch layer can fetch them ahead of the pass. A no-op without a sink or without a
-    /// prefetch-capable inner reader.
+    /// Report the byte ranges the payload pass is about to read to pre_buffer_range_callback_, so
+    /// that the prefetch layer can fetch them ahead of the pass. A no-op without a callback or
+    /// without a prefetch-capable inner reader.
     Status ReportPayloadPreBufferRanges();
 
     /// Returns the inner reader's prefetch interface, or an error when the format reader does not
@@ -178,7 +178,7 @@ class LateMaterializingFileBatchReader : public PrefetchFileBatchReader {
     std::shared_ptr<arrow::MemoryPool> arrow_pool_;
     /// Where the payload byte ranges are reported. Part of how this reader is wired into the
     /// prefetch layer rather than of a read generation, so Reset() leaves it alone.
-    PreBufferSink pre_buffer_sink_;
+    PreBufferRangeCallback pre_buffer_range_callback_;
     LatMatState state_ = kInit;
     std::shared_ptr<arrow::Schema> full_schema_;
     // projection holding only the predicate fields; nullptr when probing is not applicable

src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp

@@ -766,9 +766,9 @@ TEST_F(LateMaterializingFileBatchReaderTest, FailsOnPredicateTypeMismatch) {
 }
 
 // The payload byte ranges are only known once the probe pass has run, so they are reported through
-// the sink rather than through PreBufferRange(), and early enough to still be prefetched: before
-// the first payload batch leaves the reader.
-TEST_F(LateMaterializingFileBatchReaderTest, PayloadPreBufferRangesReportedToSink) {
+// the callback rather than through PreBufferRange(), and early enough to still be prefetched:
+// before the first payload batch leaves the reader.
+TEST_F(LateMaterializingFileBatchReaderTest, PayloadPreBufferRangesReportedToCallback) {
     auto data = BuildData({0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
     auto mock = std::make_unique<MockFileBatchReader>(data, full_type_, /*batch_size=*/3);
     // Aliased because a type carrying a comma cannot be passed to the ASSERT_OK_AND_ASSIGN macro.
@@ -781,7 +781,7 @@ TEST_F(LateMaterializingFileBatchReaderTest, PayloadPreBufferRangesReportedToSin
     int32_t batches = 0;
     std::vector<ByteRanges> reported;
     std::vector<int32_t> batches_when_reported;
-    reader->SetPreBufferSink([&](ByteRanges&& ranges) {
+    reader->SetPreBufferRangeCallback([&](ByteRanges&& ranges) {
         reported.push_back(std::move(ranges));
         batches_when_reported.push_back(batches);
     });
@@ -811,14 +811,15 @@ TEST_F(LateMaterializingFileBatchReaderTest, PayloadPreBufferRangesReportedToSin
 }
 
 // Without a predicate there is no payload pass, so there is no late range to report.
-TEST_F(LateMaterializingFileBatchReaderTest, NoSinkReportWithoutLateMaterialization) {
+TEST_F(LateMaterializingFileBatchReaderTest, NoCallbackReportWithoutLateMaterialization) {
     auto data = BuildData({0, 1, 2, 3, 4});
     auto mock = std::make_unique<MockFileBatchReader>(data, full_type_, /*batch_size=*/2);
     mock->SetPreBufferRangesPerSchema({{{0, 1024}}, {{100000, 2048}}});
     ASSERT_OK_AND_ASSIGN(auto reader, LateMaterializingFileBatchReader::Create(
                                           std::move(mock), GetArrowPool(pool_)));
     int32_t calls = 0;
-    reader->SetPreBufferSink([&calls](std::vector<std::pair<uint64_t, uint64_t>>&&) { calls++; });
+    reader->SetPreBufferRangeCallback(
+        [&calls](std::vector<std::pair<uint64_t, uint64_t>>&&) { calls++; });
     ASSERT_OK(SetReadSchema(reader.get(), arrow::schema(full_fields_), /*predicate=*/nullptr,
                             std::nullopt));
 
@@ -835,7 +836,8 @@ TEST_F(LateMaterializingFileBatchReaderTest, EmptyMatchDoesNotReportPayloadRange
     ASSERT_OK_AND_ASSIGN(auto reader, LateMaterializingFileBatchReader::Create(
                                           std::move(mock), GetArrowPool(pool_)));
     int32_t calls = 0;
-    reader->SetPreBufferSink([&calls](std::vector<std::pair<uint64_t, uint64_t>>&&) { calls++; });
+    reader->SetPreBufferRangeCallback(
+        [&calls](std::vector<std::pair<uint64_t, uint64_t>>&&) { calls++; });
     auto predicate =
         PredicateBuilder::GreaterThan(/*field_index=*/0, "k", FieldType::BIGINT, Literal(100l));
     ASSERT_OK(SetReadSchema(reader.get(), arrow::schema(full_fields_), predicate, std::nullopt));
@@ -854,7 +856,8 @@ TEST_F(LateMaterializingFileBatchReaderTest, PayloadPreBufferRangeErrorFailsRead
     ASSERT_OK_AND_ASSIGN(auto reader, LateMaterializingFileBatchReader::Create(
                                           std::move(mock), GetArrowPool(pool_)));
     int32_t calls = 0;
-    reader->SetPreBufferSink([&calls](std::vector<std::pair<uint64_t, uint64_t>>&&) { calls++; });
+    reader->SetPreBufferRangeCallback(
+        [&calls](std::vector<std::pair<uint64_t, uint64_t>>&&) { calls++; });
     auto predicate =
         PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k", FieldType::BIGINT, Literal(2l));
     ASSERT_OK(SetReadSchema(reader.get(), arrow::schema(full_fields_), predicate, std::nullopt));

src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp

@@ -327,7 +327,7 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
         // late-materialization payload pass) report them here instead of through PreBufferRange(),
         // which is asked once per read-range generation before any read. Capturing the impl raw is
         // safe: it owns readers_, and CleanUp() joins the background thread before it goes away.
-        sub_reader->SetPreBufferSink(
+        sub_reader->SetPreBufferRangeCallback(
             [impl = reader.get()](std::vector<std::pair<uint64_t, uint64_t>>&& ranges) {
                 impl->RegisterLatePreBufferRanges(std::move(ranges));
             });
@@ -366,7 +366,7 @@ PrefetchFileBatchReaderImpl::PrefetchFileBatchReaderImpl(
 
 PrefetchFileBatchReaderImpl::~PrefetchFileBatchReaderImpl() {
     (void)CleanUp();
-    ClearPreBufferSinks();
+    ClearPreBufferRangeCallbacks();
 }
 
 Status PrefetchFileBatchReaderImpl::SetReadSchema(
@@ -834,11 +834,11 @@ void PrefetchFileBatchReaderImpl::RegisterLatePreBufferRanges(
     cache_->Warmup(first_added.value().value());
 }
 
-void PrefetchFileBatchReaderImpl::ClearPreBufferSinks() {
-    // Belt and braces: the sinks capture this, and by now CleanUp() has joined the background
+void PrefetchFileBatchReaderImpl::ClearPreBufferRangeCallbacks() {
+    // Belt and braces: the callbacks capture this, and by now CleanUp() has joined the background
     // thread, so no sub-reader can report a range any more.
     for (const auto& sub_reader : readers_) {
-        sub_reader->SetPreBufferSink(nullptr);
+        sub_reader->SetPreBufferRangeCallback(nullptr);
     }
 }
 
@@ -1082,7 +1082,7 @@ void PrefetchFileBatchReaderImpl::Close() {
     for (const auto& reader : readers_) {
         reader->Close();
     }
-    ClearPreBufferSinks();
+    ClearPreBufferRangeCallbacks();
 }
 
 }  // namespace paimon

src/paimon/common/reader/prefetch_file_batch_reader_impl.h

@@ -151,9 +151,9 @@ class PrefetchFileBatchReaderImpl : public PrefetchFileBatchReader {
     /// as in WarmCacheOnce().
     void RegisterLatePreBufferRanges(std::vector<std::pair<uint64_t, uint64_t>>&& read_ranges);
 
-    /// Detaches the sinks installed on the sub-readers, so that a sub-reader outliving this reader
-    /// cannot call back into it. Called once the background thread has been joined.
-    void ClearPreBufferSinks();
+    /// Detaches the callbacks installed on the sub-readers, so that a sub-reader outliving this
+    /// reader cannot call back into it. Called once the background thread has been joined.
+    void ClearPreBufferRangeCallbacks();
     void SetReadStatus(const Status& status);
     Status GetReadStatus() const;
     Result<bool> IsEofRange(const std::pair<uint64_t, uint64_t>& read_range) const;

src/paimon/common/utils/read_ahead_cache.h

@@ -132,11 +132,10 @@ struct RangeCacheEntry {
     }
 };
 
-/// A read cache designed to hide IO latencies when reading.
-/// Prefetching strategy: When a range is read, the cache will prefetch up to
-/// `pre_buffer_range_count` additional adjacent ranges ahead of the requested offset. This helps
-/// hide I/O latency for sequential access. Example: If you read range [0, 100), and
-/// pre_buffer_range_count=2, the next two configured ranges will also be prefetched.
+/// A read cache designed to hide IO latency when reading.
+/// Prefetching strategy: a Read() (or Warmup()) at an offset publishes and fetches the registered
+/// ranges from that offset forward until their cumulative size exceeds `pre_buffer_limit`, so a
+/// sequential read runs ahead of the caller instead of fetching one range at a time.
 ///
 /// The cache never evicts: every published range stays cached until
 /// ReleaseBuffers() or Reset(). It is meant to hold the prefetched ranges of
@@ -163,19 +162,15 @@ class PAIMON_EXPORT ReadAheadCache {
 
     /// Register byte ranges to prefetch, merging them into the ranges already registered.
     ///
-    /// This is the cache's only registration entry point, and it may be called repeatedly and
-    /// concurrently with Read(): the ranges known up front - a read plan's whole extent - are
-    /// registered before any Read(), and the ranges that only become known WHILE reading - a
-    /// late-materialization payload pass learns which pages it needs only after the probe pass has
-    /// evaluated the predicate - are registered as they are discovered. Read() finds its covering
-    /// entries by walking a disjoint, offset-ordered list, so of a new range only the part no
-    /// registered range covers is added; the overlapping part is dropped, as those bytes are
-    /// already being fetched. A registration is cut into ranges of at most `range_size_limit`
-    /// bytes, one prefetch IO each, so a large pass is fetched concurrently instead of in one long
-    /// request.
-    ///
-    /// Deciding WHICH ranges to register - and in particular not reporting the ranges of a read
-    /// that has already ended - is the caller's concern; the cache registers whatever it is given.
+    /// The cache's only registration entry point; safe to call repeatedly and concurrently with
+    /// Read(). Ranges known up front are registered before any Read(); ranges that only become
+    /// known while reading (a late-materialization payload pass learns its pages after the probe
+    /// pass) are registered as they are discovered. The list stays disjoint and offset-ordered, so
+    /// only the part of a new range that no registered range covers is added - the overlap is
+    /// dropped, as those bytes are already being fetched - and each registration is cut into ranges
+    /// of at most `range_size_limit` bytes (one prefetch IO each) so a large pass is fetched
+    /// concurrently. Which ranges to register, and not reporting those of an ended read, is the
+    /// caller's concern.
     ///
     /// @param ranges The byte ranges to register.
     /// @return The offset of the first newly registered range, or nullopt when nothing was added.
@@ -195,36 +190,33 @@ class PAIMON_EXPORT ReadAheadCache {
     /// filled; false on cache miss (`dest` is left untouched).
     Result<bool> Read(const ByteRange& range, char* dest);
 
-    /// Start fetching the first batch of pending ranges immediately.
-    /// AddRanges() only registers the ranges; without Warmup() the first fetch starts
-    /// when the first Read() arrives, racing the caller's own miss fetch.
+    /// Start fetching from the first registered range forward immediately. AddRanges() only
+    /// registers the ranges; without Warmup() the first fetch starts when the first Read() arrives,
+    /// racing the caller's own miss fetch.
     void Warmup();
 
     /// Start fetching the registered ranges from `from_offset` forward, bounded by the
     /// pre-buffer limit. Warmup() is this call made from the first registered range.
     /// @param from_offset The offset to start fetching from, typically one AddRanges() returned.
     void Warmup(uint64_t from_offset);
 
-    /// Collect the counters of the Read() calls, of the block cache, of the IOs
-    /// and of the late registrations into the given metrics as counters named
-    /// after `ReadAheadCacheMetrics`. Only reads issued through Read() are
-    /// counted as hits, block hits or misses; the fetches the cache dispatches
-    /// are counted in the io counters instead.
+    /// 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, block hits or misses; the fetches the cache dispatches 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;
 
-    /// Reset the cache to its initial state, clearing all cached data and configuration.
-    ///
-    /// This method waits for all ongoing asynchronous read operations to complete,
-    /// clears all cached entries, and resets the internal state so a fresh set of ranges can be
-    /// registered again with AddRanges().
+    /// Drop the prefetched entries and zero all counters, so a fresh set of ranges can be
+    /// registered with AddRanges(). Waits for the in-flight fetches before their buffers go away.
     ///
-    /// 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.
+    /// 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 - but its counters are zeroed too. Unlike
+    /// ReleaseBuffers(), the counters do not survive.
     void Reset();
 
-    /// Release all cached buffers and pending ranges while keeping the hit/miss
+    /// Release all cached buffers and registered ranges while keeping the hit/miss
     /// counters intact.
     ///
     /// Unlike Reset(), the counters recorded by Read() remain readable through