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 7470e4c6 perf(read): warm up next data file in ConcatBatchReader (#420)
7470e4c6 is described below
commit 7470e4c62c8970683474e5fdc0e5fceb8f138857
Author: Nicholas Jiang <[email protected]>
AuthorDate: Fri Oct 9 15:40:47 2026 +0700
perf(read): warm up next data file in ConcatBatchReader (#420)
---
docs/source/user_guide/prefetch.rst | 48 ++++++++++
src/paimon/common/reader/concat_batch_reader.cpp | 25 ++++-
src/paimon/common/reader/concat_batch_reader.h | 12 +++
.../common/reader/concat_batch_reader_test.cpp | 102 +++++++++++++++++++++
4 files changed, 186 insertions(+), 1 deletion(-)
diff --git a/docs/source/user_guide/prefetch.rst
b/docs/source/user_guide/prefetch.rst
index 4e601219..e8a3a348 100644
--- a/docs/source/user_guide/prefetch.rst
+++ b/docs/source/user_guide/prefetch.rst
@@ -70,3 +70,51 @@ Stripe metadata, the selected columns, and the configured
natural read size.
Prefetch is enabled when the estimated compressed size of the selected columns
exceeds the configured threshold; otherwise the reader uses the normal
single-reader path.
+
+Warmup
+======
+
+Prefetching parallelizes the reads inside one file. When the data files of a
+split are read one after another, warmup also overlaps the files with each
+other: while one file is being consumed, the reader already starts the first
+read of the next one, instead of waiting until the current file reaches its
+end. Warmup looks one file ahead of the file being read, and merge-on-read with
+the default ``loser-tree`` sort engine additionally warms the first file of
+every sorted run before merging. It applies to append-table reads and append
+compaction as well, but does not cross split boundaries.
+
+How far the next file is prepared is set with
+``ReadContextBuilder::SetWarmupLevel()``. Append compaction builds its own read
+context and always uses the default level.
+
+- ``WarmupLevel::NONE``: do not warm up.
+- ``WarmupLevel::RAW`` (default): fetch the next file's raw bytes through the
+ read-ahead cache without decoding them. It has no effect when the read-ahead
+ cache is disabled with ``ReadContextBuilder::SetReadAheadCacheEnabled()``.
+- ``WarmupLevel::DECODED``: also start decoding the next file, which hides the
+ most latency but uses the most memory.
+
+Warmup goes through the prefetch reader, so it only applies to Parquet and ORC
+files that are actually read with prefetching, and enabling prefetch alone does
+not guarantee that. With the adaptive strategy, a file whose first read range
+holds more batches than one prefetch queue can buffer is read without
+prefetching, and so without warmup; the ``prefetch.adaptive-disabled-count``
+metric counts how often this happens. Append compaction uses a prefetch queue
+of three batches, so it only warms files whose first read range spans no more
+than three full batches, such as files with small row groups.
+
+``WarmupLevel::RAW`` issues its fetches from the reading thread through the
+file system's asynchronous read, so it only overlaps I/O when that read is
+really asynchronous. The local file system serves it synchronously, so there
+``RAW`` warmup reads the next file on the reading thread before the current
+batch is returned, adding latency instead of hiding it. On such file systems,
+prefer ``WarmupLevel::DECODED``, which fetches on the background decode thread,
+or ``WarmupLevel::NONE``.
+
+Warmup preserves the returned rows and their order, but it costs memory: under
+``WarmupLevel::RAW`` the next file may hold up to the read-ahead cache's
+pre-buffer limit (256 MiB by default, see ``CacheConfig::SetPreBufferLimit()``)
+on top of the file being read. It may also fetch data from a file that the
+reader never consumes: a read that stops early, for example because of a LIMIT,
+may already have warmed the next file, and closing it waits for that file's
+in-flight fetches.
diff --git a/src/paimon/common/reader/concat_batch_reader.cpp
b/src/paimon/common/reader/concat_batch_reader.cpp
index 5aea88e3..167fdc5f 100644
--- a/src/paimon/common/reader/concat_batch_reader.cpp
+++ b/src/paimon/common/reader/concat_batch_reader.cpp
@@ -19,12 +19,14 @@
#include "paimon/common/reader/concat_batch_reader.h"
+#include <algorithm>
#include <utility>
#include "arrow/c/abi.h"
#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/reader/reader_utils.h"
#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/reader/file_batch_reader.h"
namespace paimon {
class MemoryPool;
@@ -34,7 +36,15 @@
ConcatBatchReader::ConcatBatchReader(std::vector<std::unique_ptr<BatchReader>>&&
: arrow_pool_(arrow_pool),
finished_reader_metrics_(std::make_shared<MetricsImpl>()),
readers_(std::move(readers)),
- current_(0) {}
+ file_readers_(readers_.size(), nullptr),
+ current_(0) {
+ // Non-file readers keep their null slots so the warmup window stays
aligned with readers_.
+ for (size_t i = 0; i < readers_.size(); i++) {
+ if (auto* file_reader =
dynamic_cast<FileBatchReader*>(readers_[i].get())) {
+ file_readers_[i] = file_reader;
+ }
+ }
+}
Result<BatchReader::ReadBatch> ConcatBatchReader::NextBatch() {
PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch_with_bitmap,
@@ -66,10 +76,23 @@ void ConcatBatchReader::CloseAndReleaseReader(size_t
reader_index) {
reader->Close();
finished_reader_metrics_->Merge(reader->GetReaderMetrics());
reader.reset();
+ file_readers_[reader_index] = nullptr;
+}
+
+void ConcatBatchReader::WarmupRange(size_t idx, size_t count) {
+ const size_t end = std::min(idx + count, file_readers_.size());
+ for (size_t i = idx; i < end; i++) {
+ if (file_readers_[i] != nullptr) {
+ file_readers_[i]->Warmup();
+ }
+ }
}
Result<BatchReader::ReadBatchWithBitmap>
ConcatBatchReader::NextBatchWithBitmap() {
while (current_ < readers_.size()) {
+ // Lookahead: when the file system reads asynchronously, the next
file's first read is paid
+ // while this file is still being consumed, instead of serially after
its EOF.
+ WarmupRange(current_, 1 + kWarmupLookahead);
auto& current_reader = readers_[current_];
PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap result,
current_reader->NextBatchWithBitmap());
diff --git a/src/paimon/common/reader/concat_batch_reader.h
b/src/paimon/common/reader/concat_batch_reader.h
index 2c889d44..031e6187 100644
--- a/src/paimon/common/reader/concat_batch_reader.h
+++ b/src/paimon/common/reader/concat_batch_reader.h
@@ -29,6 +29,7 @@
#include "paimon/result.h"
namespace paimon {
+class FileBatchReader;
class MemoryPool;
/// This reader is to concatenate a list of BatchReaders and read them
sequentially. The input list
@@ -44,11 +45,22 @@ class ConcatBatchReader : public BatchReader {
std::shared_ptr<Metrics> GetReaderMetrics() const override;
private:
+ /// How many child readers ahead of the one being consumed get their first
read started. Each
+ /// warm reader holds its own prefetched data, so one ahead is a
conservative trade-off between
+ /// overlapped I/O and extra memory.
+ static constexpr size_t kWarmupLookahead = 1;
+
+ /// Warms up \p count readers starting at \p idx, stopping at the end of
the list. Warmup() is
+ /// idempotent, so this is safe to call on every batch.
+ void WarmupRange(size_t idx, size_t count);
void CloseAndReleaseReader(size_t reader_index);
std::shared_ptr<arrow::MemoryPool> arrow_pool_;
std::shared_ptr<Metrics> finished_reader_metrics_;
std::vector<std::unique_ptr<BatchReader>> readers_;
+ // readers_[i] as a FileBatchReader, or nullptr when it is not one or has
been released.
+ // Warmup() lives on FileBatchReader, so the cast is done once instead of
on every batch.
+ std::vector<FileBatchReader*> file_readers_;
size_t current_;
};
} // namespace paimon
diff --git a/src/paimon/common/reader/concat_batch_reader_test.cpp
b/src/paimon/common/reader/concat_batch_reader_test.cpp
index 3cc66e08..4d3ff81c 100644
--- a/src/paimon/common/reader/concat_batch_reader_test.cpp
+++ b/src/paimon/common/reader/concat_batch_reader_test.cpp
@@ -98,6 +98,43 @@ class FixedMetricsBatchReader : public BatchReader {
std::shared_ptr<Metrics> metrics_;
};
+namespace {
+/// Logs its Warmup() and NextBatchWithBitmap() calls as "warmup <name>" and
"read <name>" into a
+/// log the test owns, because ConcatBatchReader destroys a child once it
reaches EOF and the
+/// child's own state goes with it.
+class EventLoggingFileBatchReader : public MockFileBatchReader {
+ public:
+ EventLoggingFileBatchReader(const std::shared_ptr<arrow::Array>& data,
const std::string& name,
+ std::vector<std::string>* events)
+ : MockFileBatchReader(data, data->type(), /*read_batch_size=*/1),
+ name_(name),
+ events_(events) {}
+
+ void Warmup() override {
+ events_->push_back("warmup " + name_);
+ MockFileBatchReader::Warmup();
+ }
+
+ Result<ReadBatchWithBitmap> NextBatchWithBitmap() override {
+ events_->push_back("read " + name_);
+ return MockFileBatchReader::NextBatchWithBitmap();
+ }
+
+ private:
+ std::string name_;
+ std::vector<std::string>* events_;
+};
+
+int64_t CountEvent(const std::vector<std::string>& events, const std::string&
event) {
+ return std::count(events.begin(), events.end(), event);
+}
+
+/// Position of the first \p event in \p events, or events.size() when it
never happened.
+size_t FirstEvent(const std::vector<std::string>& events, const std::string&
event) {
+ return static_cast<size_t>(std::find(events.begin(), events.end(), event)
- events.begin());
+}
+} // namespace
+
class ConcatBatchReaderTest : public ::testing::Test {
void SetUp() override {
pool_ = GetDefaultPool();
@@ -283,4 +320,69 @@ TEST_F(ConcatBatchReaderTest,
TestCollectMetricsAfterReleasingReaders) {
ASSERT_EQ(io_count, 5);
}
+TEST_F(ConcatBatchReaderTest, TestWarmupLooksOneReaderAhead) {
+ std::shared_ptr<arrow::DataType> type = arrow::struct_({arrow::field("f1",
arrow::int32())});
+ std::vector<std::string> events;
+ std::vector<std::unique_ptr<BatchReader>> readers;
+ readers.push_back(std::make_unique<EventLoggingFileBatchReader>(
+ arrow::ipc::internal::json::ArrayFromJSON(type, "[[1],
[2]]").ValueOrDie(), "a", &events));
+ readers.push_back(std::make_unique<EventLoggingFileBatchReader>(
+ arrow::ipc::internal::json::ArrayFromJSON(type, "[[3]]").ValueOrDie(),
"b", &events));
+ readers.push_back(std::make_unique<EventLoggingFileBatchReader>(
+ arrow::ipc::internal::json::ArrayFromJSON(type, "[[4]]").ValueOrDie(),
"c", &events));
+ ConcatBatchReader reader(std::move(readers), GetArrowPool(pool_));
+ ASSERT_TRUE(events.empty());
+
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap batch,
reader.NextBatchWithBitmap());
+ ASSERT_FALSE(BatchReader::IsEofBatch(batch));
+ ReaderUtils::ReleaseReadBatch(std::move(batch.first));
+ ASSERT_LT(FirstEvent(events, "warmup a"), FirstEvent(events, "read a"));
+ ASSERT_LT(FirstEvent(events, "warmup b"), FirstEvent(events, "read a"));
+ ASSERT_EQ(CountEvent(events, "warmup c"), 0);
+
+ // Still inside the first file, so the lookahead does not move on.
+ ASSERT_OK_AND_ASSIGN(batch, reader.NextBatchWithBitmap());
+ ASSERT_FALSE(BatchReader::IsEofBatch(batch));
+ ReaderUtils::ReleaseReadBatch(std::move(batch.first));
+ ASSERT_EQ(CountEvent(events, "warmup c"), 0);
+
+ ASSERT_OK_AND_ASSIGN(batch, reader.NextBatchWithBitmap());
+ ASSERT_FALSE(BatchReader::IsEofBatch(batch));
+ ReaderUtils::ReleaseReadBatch(std::move(batch.first));
+ ASSERT_LT(FirstEvent(events, "warmup c"), FirstEvent(events, "read b"));
+ reader.Close();
+}
+
+TEST_F(ConcatBatchReaderTest, TestWarmupSkipsReaderThatIsNotFileBatchReader) {
+ std::shared_ptr<arrow::DataType> type = arrow::struct_({arrow::field("f1",
arrow::int32())});
+ std::vector<std::string> events;
+ std::vector<std::unique_ptr<BatchReader>> readers;
+ readers.push_back(std::make_unique<EventLoggingFileBatchReader>(
+ arrow::ipc::internal::json::ArrayFromJSON(type, "[[1],
[2]]").ValueOrDie(), "a", &events));
+ readers.push_back(std::make_unique<LifetimeTrackingBatchReader>(
+ std::make_unique<EventLoggingFileBatchReader>(
+ arrow::ipc::internal::json::ArrayFromJSON(type,
"[[3]]").ValueOrDie(), "b", &events),
+ /*lifetime=*/nullptr));
+ readers.push_back(std::make_unique<EventLoggingFileBatchReader>(
+ arrow::ipc::internal::json::ArrayFromJSON(type, "[[4]]").ValueOrDie(),
"c", &events));
+ auto concat_reader =
+ std::make_unique<ConcatBatchReader>(std::move(readers),
GetArrowPool(pool_));
+
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap batch,
+ concat_reader->NextBatchWithBitmap());
+ ASSERT_FALSE(BatchReader::IsEofBatch(batch));
+ ReaderUtils::ReleaseReadBatch(std::move(batch.first));
+ // The lookahead window counts every child, so it does not reach past the
skipped one.
+ ASSERT_EQ(CountEvent(events, "warmup c"), 0);
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+
ReadResultCollector::CollectResult(concat_reader.get()));
+ std::shared_ptr<arrow::Array> expected =
+ arrow::ipc::internal::json::ArrayFromJSON(type, "[[2], [3],
[4]]").ValueOrDie();
+
ASSERT_TRUE(std::make_shared<arrow::ChunkedArray>(expected)->Equals(result))
+ << result->ToString();
+ ASSERT_EQ(CountEvent(events, "warmup b"), 0);
+ ASSERT_LT(FirstEvent(events, "warmup c"), FirstEvent(events, "read b"));
+}
+
} // namespace paimon::test
