fix

Author: lucasfang

include/paimon/reader/prefetch_file_batch_reader.h

@@ -148,7 +148,7 @@ class PAIMON_EXPORT PrefetchFileBatchReader : public FileBatchReader {
     /// 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*/) {}
+    virtual void SetPreBufferSink(PreBufferSink sink) {}
 };
 
 }  // namespace paimon

include/paimon/utils/prefetch_cache_config.h

@@ -59,7 +59,7 @@ class PAIMON_EXPORT CacheConfig {
     }
 
     /// Returns the maximum gap size (in bytes) considered mergeable between
-    /// adjacent ranges. Defaults to 8 KiB.
+    /// adjacent ranges. Defaults to 512 KiB.
     uint64_t GetHoleSizeLimit() const {
         return hole_size_limit_;
     }
@@ -103,6 +103,31 @@ class PAIMON_EXPORT CacheConfig {
         block_cache_limit_ = block_cache_limit;
     }
 
+    /// Returns the alignment (in bytes) the adaptive range size is rounded up to, which is also
+    /// the smallest size a range is cut to. Defaults to 4 MiB.
+    uint64_t GetRangeSplitAlignment() const {
+        return range_split_alignment_;
+    }
+
+    /// Sets the alignment the adaptive range size is rounded up to. Zero turns the adaptive
+    /// sizing off, so the ranges are cut at the configured size limit instead.
+    void SetRangeSplitAlignment(uint64_t range_split_alignment) {
+        range_split_alignment_ = range_split_alignment;
+    }
+
+    /// Returns the number of requests the adaptive range size aims to spread one round of
+    /// registered ranges over. Defaults to 10.
+    uint64_t GetRangeSplitConcurrency() const {
+        return range_split_concurrency_;
+    }
+
+    /// Sets the number of requests the adaptive range size aims to spread one round of ranges
+    /// over. Zero turns the adaptive sizing off, so the ranges are cut at the configured size
+    /// limit instead.
+    void SetRangeSplitConcurrency(uint64_t range_split_concurrency) {
+        range_split_concurrency_ = range_split_concurrency;
+    }
+
  private:
     // The defaults are aligned with the reader's request granularity and with
     // realistic data file sizes:
@@ -117,11 +142,29 @@ class PAIMON_EXPORT CacheConfig {
     //   at Init: those are fetched only just before they are read, and one range
     //   is one request, so they are cut smaller than range_size_limit to be
     //   fetched concurrently rather than in one long request. A read spanning
-    //   several of them is still served, as they are adjacent.
+    //   several of them is still served, as they are adjacent. It is an upper
+    //   bound only: the size such a round is actually cut at is derived from the
+    //   bytes it registers, see range_split_alignment below.
+    // - hole_size_limit trades bytes against requests: coalescing across a gap
+    //   reads the gap too, but saves a request, and on remote storage a request
+    //   costs a round trip whatever its size. The limit is therefore well above
+    //   the page-sized gaps a filtered read leaves between the pages it keeps,
+    //   which would otherwise each cost a request of their own.
     uint64_t range_size_limit_ = 32 * 1024 * 1024;
     uint64_t late_range_size_limit_ = 8 * 1024 * 1024;
-    uint64_t hole_size_limit_ = 8 * 1024;
+    uint64_t hole_size_limit_ = 512 * 1024;
     uint64_t pre_buffer_limit_ = 256 * 1024 * 1024;
+    // A fixed size limit cuts a small round into fewer requests than could be
+    // fetched at once, leaving the storage idle while each of them runs. So the
+    // size a round is cut at is derived from the round instead: its bytes are
+    // spread over range_split_concurrency requests, rounded up to
+    // range_split_alignment, and kept within the size limit above. A round
+    // smaller than alignment * concurrency is therefore cut finer than that
+    // limit and fetched in a single wave, while a larger one still stops at it.
+    // The alignment is also the floor, so the derivation never cuts a round into
+    // requests too small to amortize their own round trip.
+    uint64_t range_split_alignment_ = 4 * 1024 * 1024;
+    uint64_t range_split_concurrency_ = 10;
     // 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

src/paimon/common/utils/byte_range_combiner.cpp

@@ -29,6 +29,37 @@
 
 namespace paimon {
 
+namespace {
+
+/// The size to cut `total_bytes` of ranges at so that they are spread over
+/// `target_concurrency` requests, rounded up to `alignment` and kept in
+/// (hole_size_limit, range_size_limit].
+uint64_t AdaptiveRangeSizeLimit(uint64_t total_bytes, uint64_t target_concurrency,
+                                uint64_t alignment, uint64_t hole_size_limit,
+                                uint64_t range_size_limit) {
+    // Leave the limit alone when the derivation is off, and when it is not larger than the hole
+    // size: CoalesceByteRanges() rejects such a pair, and it must report that as it always did
+    // rather than have a derived limit paper over it.
+    if (alignment == 0 || target_concurrency == 0 || range_size_limit <= hole_size_limit) {
+        return range_size_limit;
+    }
+    const uint64_t per_request =
+        total_bytes / target_concurrency + (total_bytes % target_concurrency == 0 ? 0 : 1);
+    // One alignment unit is the floor, and the limit must stay above the hole size for the
+    // reason above. hole_size_limit + 1 cannot overflow: it is below range_size_limit here.
+    uint64_t limit = std::max({alignment, per_request, hole_size_limit + 1});
+    if (const uint64_t remainder = limit % alignment; remainder != 0) {
+        const uint64_t padding = alignment - remainder;
+        if (limit > std::numeric_limits<uint64_t>::max() - padding) {
+            return range_size_limit;
+        }
+        limit += padding;
+    }
+    return std::min(limit, range_size_limit);
+}
+
+}  // namespace
+
 Result<std::vector<ByteRange>> ByteRangeCombiner::CoalesceByteRanges(
     std::vector<ByteRange>&& ranges, uint64_t hole_size_limit, uint64_t range_size_limit) {
     if (range_size_limit <= hole_size_limit) {
@@ -137,4 +168,20 @@ Result<std::vector<ByteRange>> ByteRangeCombiner::CoalesceByteRanges(
     return coalesced;
 }
 
+Result<std::vector<ByteRange>> ByteRangeCombiner::CoalesceByteRangesAdaptive(
+    std::vector<ByteRange>&& ranges, uint64_t hole_size_limit, uint64_t range_size_limit,
+    uint64_t alignment, uint64_t target_concurrency) {
+    uint64_t total_bytes = 0;
+    for (const auto& range : ranges) {
+        // Saturate rather than wrap. CoalesceByteRanges() rejects the ranges that can add up
+        // this far, but only after the limit below has been derived from them.
+        total_bytes = range.length > std::numeric_limits<uint64_t>::max() - total_bytes
+                          ? std::numeric_limits<uint64_t>::max()
+                          : total_bytes + range.length;
+    }
+    return CoalesceByteRanges(std::move(ranges), hole_size_limit,
+                              AdaptiveRangeSizeLimit(total_bytes, target_concurrency, alignment,
+                                                     hole_size_limit, range_size_limit));
+}
+
 }  // namespace paimon

src/paimon/common/utils/byte_range_combiner.h

@@ -32,6 +32,20 @@ struct ByteRangeCombiner {
     static Result<std::vector<ByteRange>> CoalesceByteRanges(std::vector<ByteRange>&& ranges,
                                                              uint64_t hole_size_limit,
                                                              uint64_t range_size_limit);
+
+    /// Coalesces `ranges` like CoalesceByteRanges() does, except that the size the ranges are
+    /// cut at is derived from how many bytes they cover instead of being fixed: the bytes are
+    /// spread over `target_concurrency` ranges, the result is rounded up to a multiple of
+    /// `alignment`, and it is kept within `range_size_limit`. So a small batch of ranges is cut
+    /// into as many ranges as can be fetched at once rather than into fewer, longer ones, while
+    /// a batch large enough to reach `range_size_limit` is cut exactly as CoalesceByteRanges()
+    /// would cut it.
+    ///
+    /// A zero `alignment` or `target_concurrency` turns the derivation off, falling back to
+    /// `range_size_limit`.
+    static Result<std::vector<ByteRange>> CoalesceByteRangesAdaptive(
+        std::vector<ByteRange>&& ranges, uint64_t hole_size_limit, uint64_t range_size_limit,
+        uint64_t alignment, uint64_t target_concurrency);
 };
 
 }  // namespace paimon

src/paimon/common/utils/byte_range_combiner_test.cpp

@@ -99,4 +99,53 @@ TEST(ByteRangeCombinerTest, TestRejectsRangesBeyondInt64Bound) {
     check_invalid({{100, 10}, {0, std::numeric_limits<uint64_t>::max()}});
 }
 
+// The adaptive variant derives the size the ranges are cut at from the bytes they cover, so a
+// batch too small to fill the concurrency on its own is cut finer than the fixed limit cuts it.
+TEST(ByteRangeCombinerTest, TestAdaptiveRangeSizeLimit) {
+    auto check = [](std::vector<ByteRange> ranges, uint64_t alignment, uint64_t target_concurrency,
+                    std::vector<ByteRange> expected) -> void {
+        ASSERT_OK_AND_ASSIGN(auto coalesced,
+                             ByteRangeCombiner::CoalesceByteRangesAdaptive(
+                                 std::move(ranges), /*hole_size_limit=*/1,
+                                 /*range_size_limit=*/8, alignment, target_concurrency));
+        ASSERT_EQ(coalesced, expected);
+    };
+
+    // 20 bytes over 10 requests is 2 per request, which the alignment lifts to one 4 byte
+    // range each: cut at 4 instead of at the limit of 8.
+    check({{0, 20}}, /*alignment=*/4, /*target_concurrency=*/10,
+          {{0, 4}, {4, 4}, {8, 4}, {12, 4}, {16, 4}});
+    // The same bytes cut at the fixed limit, for comparison: 3 ranges rather than 5.
+    check({{0, 20}}, /*alignment=*/0, /*target_concurrency=*/10, {{0, 8}, {8, 8}, {16, 4}});
+    check({{0, 20}}, /*alignment=*/4, /*target_concurrency=*/0, {{0, 8}, {8, 8}, {16, 4}});
+    // 30 bytes over 3 requests is 10 each, which rounds up past the limit, so the limit wins
+    // and the batch is cut exactly as the non-adaptive variant cuts it.
+    check({{0, 30}}, /*alignment=*/4, /*target_concurrency=*/3, {{0, 8}, {8, 8}, {16, 8}, {24, 6}});
+    // The alignment is the floor: 10 bytes over 10 requests is 1 each, not cut into 1 byte
+    // requests that could never amortize their own round trip.
+    check({{0, 10}}, /*alignment=*/4, /*target_concurrency=*/10, {{0, 4}, {4, 4}, {8, 2}});
+    // A derived limit still coalesces across the holes that fit within it, and still bounds the
+    // coalesced result: the third range would take it past 4 bytes, so it starts a new range.
+    check({{0, 1}, {2, 1}}, /*alignment=*/4, /*target_concurrency=*/10, {{0, 3}});
+    check({{0, 1}, {2, 1}, {4, 1}}, /*alignment=*/4, /*target_concurrency=*/10, {{0, 3}, {4, 1}});
+    check({}, /*alignment=*/4, /*target_concurrency=*/10, {});
+}
+
+// CoalesceByteRanges() rejects a size limit that does not exceed the hole size, so the derived
+// limit must stay above it however small the batch is.
+TEST(ByteRangeCombinerTest, TestAdaptiveRangeSizeLimitStaysAboveHoleSize) {
+    // 10 bytes over 10 requests would derive 4, below the hole size of 5.
+    ASSERT_OK_AND_ASSIGN(auto coalesced, ByteRangeCombiner::CoalesceByteRangesAdaptive(
+                                             {{0, 10}}, /*hole_size_limit=*/5,
+                                             /*range_size_limit=*/8, /*alignment=*/4,
+                                             /*target_concurrency=*/10));
+    ASSERT_EQ(coalesced, (std::vector<ByteRange>{{0, 8}, {8, 2}}));
+
+    // A limit that cannot exceed the hole size is still reported, not papered over.
+    ASSERT_NOK_WITH_MSG(ByteRangeCombiner::CoalesceByteRangesAdaptive(
+                            {{0, 10}}, /*hole_size_limit=*/8, /*range_size_limit=*/8,
+                            /*alignment=*/4, /*target_concurrency=*/10),
+                        "should be larger than hole size limit");
+}
+
 }  // namespace paimon::test

src/paimon/common/utils/read_ahead_cache.cpp

@@ -191,8 +191,9 @@ Result<std::optional<uint64_t>> ReadAheadCache::Impl::AddRanges(std::vector<Byte
                                                                 uint64_t expected_round) {
     PAIMON_ASSIGN_OR_RAISE(
         std::vector<ByteRange> new_ranges,
-        ByteRangeCombiner::CoalesceByteRanges(std::move(ranges), config_.GetHoleSizeLimit(),
-                                              config_.GetLateRangeSizeLimit()));
+        ByteRangeCombiner::CoalesceByteRangesAdaptive(
+            std::move(ranges), config_.GetHoleSizeLimit(), config_.GetLateRangeSizeLimit(),
+            config_.GetRangeSplitAlignment(), config_.GetRangeSplitConcurrency()));
     for (const auto& new_range : new_ranges) {
         PAIMON_RETURN_NOT_OK(ValidateValueInRange<int64_t>(new_range.offset, "range offset"));
         PAIMON_RETURN_NOT_OK(ValidateValueInRange<int64_t>(new_range.length, "range length"));