This is an automated email from the ASF dual-hosted git repository.
sollhui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8d88f1cc6f7 [perf](load) introduce sequential prefetch for
FileScannerV2 CSV reads (#69716)
8d88f1cc6f7 is described below
commit 8d88f1cc6f738a331ecfe024a2f1560c770068bb
Author: hui lai <[email protected]>
AuthorDate: Thu Oct 29 15:20:15 2026 +0800
[perf](load) introduce sequential prefetch for FileScannerV2 CSV reads
(#69716)
### What problem does this PR solve?
Issue Number: None
Related PR: #69333
Problem Summary:
After CSV ingestion switched to FileScannerV2, `DelimitedTextReader`
opened remote files directly through `FileFactory`, bypassing the
sequential prefetch used by the original CSV reader. S3 reads therefore
block the scanner instead of overlapping remote I/O with CSV parsing.
Restore `DelegateReader::create_file_reader(SEQUENTIAL)` in the shared
CSV/Text reader, passing the shared IOContext and adjusted split range.
Unknown-size first splits use an end sentinel bounded by the physical
file size. Tracing remains outside the prefetch layer, and shared
stream-load pipes retain their existing ownership and cancellation
behavior.
To safely integrate asynchronous reads with V2 scanner statistics:
- Each prefetch buffer owns its cache statistics and an IOContext
snapshot. The consumer merges completed statistics exactly once before
consuming, reusing or closing the buffer. Workers never update the
scanner's cache-statistics object.
- Close drains active reads even after a foreground timeout, keeping
borrowed scanner state alive until I/O completes. Queued tasks observe
CLOSED and skip reading. Cleanup can consequently take longer than
`buffered_reader_read_timeout_ms` while an underlying request is
outstanding.
- The ring modulus uses allocated capacity: a 6 MiB setting allocates
one 4 MiB buffer and wraps at 4 MiB. Settings below 4 MiB use one
buffer; the 16 MiB default is unchanged.
- CSV/Text filesystem readers close before collecting prefetch profile
counters. Repeated close does not publish counters twice.
### Performance
For the TPC-H dataset load benchmark, elapsed time decreased from
**2617.33 s** to **1696.06 s**, saving **921.27 s**. This is a **35.2%
reduction in load time**, or approximately **1.54x speedup**, for the
measured workload.
| Dataset | Before | After | Load time reduction | Speedup |
| --- | ---: | ---: | ---: | ---: |
| TPC-H | 2617.33 s | 1696.06 s | 35.2% | 1.54x |
### Release note
Restore asynchronous S3 prefetch for FileScannerV2 CSV and text reads,
with safe cache-statistics aggregation and cleanup, consistent ring
capacity and populated prefetch profiles.
### Check List (For Author)
- Test:
- Added S3 mock coverage for whole-file, mid-row split and unknown-size
reads; checks row values, outer tracing, background S3 reads, prefetch
profile publication and repeated close.
- Added buffered-reader tests for four concurrent workers with isolated
cache statistics, partial consumption, slot reuse, unconsumed-buffer
collection, successful reads finishing after foreground timeout,
explicit close/destruction, and exact bytes across 0/1/6/10/16 MiB ring
settings from zero/nonzero offsets. Added a shared stream-pipe ownership
test.
- Passed Clang 16 syntax checking of the complete
`buffered_reader_test.cpp` translation unit against repository and
installed third-party headers with `-Werror -Wunused-member-function`.
- Passed repository clang-format 16 formatting/check scripts,
`build-support/check-build-hygiene.sh`, and `git diff --check`.
- Full BE build and runtime tests are **not verified locally**:
`./run-be-ut.sh --run --filter='BufferedReaderTest.*:CsvV2ReaderTest.*'
-j 8` stops at JDK initialization (`Failed setting boot class path`).
Separate production/CSV syntax attempts encounter installed C++ library
and stale generated-header incompatibilities. No compilation database is
available for clang-tidy. Runtime and sanitizer validation remain
pending CI.
- The S3 mock test disables file cache; the concurrent statistics test
exercises the actual prefetch implementation with a mock reader updating
`FileCacheStatistics`. No cached-S3 end-to-end runtime result is
claimed.
- TPC-H load benchmark: elapsed time decreased from 2617.33 s to 1696.06
s (35.2% less load time; approximately 1.54x speedup).
- Behavior changed:
- [x] Yes. Eligible S3 CSV/Text readers use sequential prefetch. Shared
prefetch readers isolate cache counters, drain outstanding reads during
cleanup and normalize ring capacity. CSV/Text publishes prefetch
profiles on close.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../delimited_text/delimited_text_reader.cpp | 18 +-
be/src/io/fs/buffered_reader.cpp | 43 +++--
be/src/io/fs/buffered_reader.h | 8 +
.../format_v2/delimited_text/csv_reader_test.cpp | 164 +++++++++++++++++++
be/test/io/fs/buffered_reader_test.cpp | 181 ++++++++++++++++++---
5 files changed, 373 insertions(+), 41 deletions(-)
diff --git a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
index 049e7265639..7aa9dff1edf 100644
--- a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
+++ b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
@@ -37,6 +37,7 @@
#include "format_v2/column_mapper.h"
#include "format_v2/materialized_reader_util.h"
#include "io/file_factory.h"
+#include "io/fs/buffered_reader.h"
#include "io/fs/tracing_file_reader.h"
#include "runtime/descriptors.h"
#include "runtime/file_scan_profile.h"
@@ -486,6 +487,13 @@ Status DelimitedTextReader::close() {
_line_reader.reset();
}
_decompressor.reset();
+ // Stream pipes are shared with the load context; preserve their existing
ownership and
+ // cancellation behavior. Only filesystem readers enter the sequential
prefetch path.
+ if (_file_reader != nullptr && _scan_params->file_type !=
TFileType::FILE_STREAM) {
+ // Drain prefetch workers before publishing their counters or
releasing IOContext refs.
+ RETURN_IF_ERROR(_file_reader->close());
+ _file_reader->collect_profile_before_close();
+ }
_file_reader.reset();
_tracing_file_reader.reset();
_requested_columns.clear();
@@ -605,8 +613,14 @@ Status DelimitedTextReader::_open_file() {
} else {
auto reader_options =
FileFactory::get_reader_options(_runtime_state->query_options(),
*_file_description);
- auto file_reader = DORIS_TRY(FileFactory::create_file_reader(
- *_system_properties, *_file_description, reader_options,
_profile));
+ // Include the prefix used to align non-first splits. Unknown-size
first splits read to
+ // EOF; PrefetchBufferedReader bounds this range by the opened file's
physical size.
+ const auto prefetch_end = _size < 0 ?
std::numeric_limits<size_t>::max()
+ : cast_set<size_t>(_start_offset +
_size);
+ auto file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, *_system_properties, *_file_description,
reader_options,
+ io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
+ io::PrefetchRange(_start_offset, prefetch_end)));
_file_reader = _io_ctx && _io_ctx->file_reader_stats
?
std::make_shared<io::TracingFileReader>(std::move(file_reader),
_io_ctx->file_reader_stats)
diff --git a/be/src/io/fs/buffered_reader.cpp b/be/src/io/fs/buffered_reader.cpp
index 599507ad751..f86a681222f 100644
--- a/be/src/io/fs/buffered_reader.cpp
+++ b/be/src/io/fs/buffered_reader.cpp
@@ -413,6 +413,13 @@ void PrefetchBuffer::reset_offset(size_t offset) {
_prefetched.notify_all();
return;
}
+ _merge_file_cache_stats();
+ // Snapshot on the consumer thread, not concurrently with scanner
counter updates.
+ // Cached readers get per-buffer counters; uncached readers keep a
null stats pointer.
+ _prefetch_io_ctx = *_io_ctx;
+ if (_io_ctx->file_cache_stats != nullptr) {
+ _prefetch_io_ctx.file_cache_stats = &_file_cache_stats;
+ }
_buffer_status = BufferStatus::RESET;
_offset = offset;
_prefetched.notify_all();
@@ -470,7 +477,7 @@ void PrefetchBuffer::prefetch_buffer() {
{
SCOPED_RAW_TIMER(&_statis.read_time);
- s = _reader->read_at(_offset, Slice {_buf.data(), buf_size}, &_len,
_io_ctx);
+ s = _reader->read_at(_offset, Slice {_buf.data(), buf_size}, &_len,
&_prefetch_io_ctx);
}
if (UNLIKELY(s.ok() && buf_size != _len)) {
// This indicates that the data size returned by S3 object storage is
smaller than what we requested,
@@ -497,6 +504,7 @@ void PrefetchBuffer::prefetch_buffer() {
_offset, s.to_string());
_prefetch_status = std::move(s);
}
+ _has_unmerged_cache_stats = _prefetch_io_ctx.file_cache_stats != nullptr;
_buffer_status = BufferStatus::PREFETCHED;
_prefetched.notify_all();
// eof would come up with len == 0, it would be handled by read_buffer
@@ -582,6 +590,7 @@ Status PrefetchBuffer::read_buffer(size_t off, const char*
out, size_t buf_len,
if (UNLIKELY(BufferStatus::CLOSED == _buffer_status)) {
return Status::OK();
}
+ _merge_file_cache_stats();
}
RETURN_IF_ERROR(_prefetch_status);
// there is only parquet would do not sequence read
@@ -612,15 +621,23 @@ Status PrefetchBuffer::read_buffer(size_t off, const
char* out, size_t buf_len,
return Status::OK();
}
+void PrefetchBuffer::_merge_file_cache_stats() {
+ // Called only by the consumer with _lock held and no pending read. A
buffer may be
+ // consumed in several slices; publish each completed request exactly once.
+ if (_has_unmerged_cache_stats) {
+ _io_ctx->file_cache_stats->merge_from(_file_cache_stats);
+ _file_cache_stats = {};
+ _has_unmerged_cache_stats = false;
+ }
+}
+
void PrefetchBuffer::close() {
std::unique_lock lck {_lock};
- // in case _reader still tries to write to the buf after we close the
buffer
- if (!_prefetched.wait_for(lck,
-
std::chrono::milliseconds(config::buffered_reader_read_timeout_ms),
- [this]() { return _buffer_status !=
BufferStatus::PENDING; })) {
- _prefetch_status = Status::TimedOut("time out when close prefetch
buffer");
- return;
- }
+ // A foreground read timeout does not cancel the underlying I/O. Wait for
it to finish
+ // before releasing scanner-owned IOContext pointees, including cache
statistics and
+ // the cache-write limiter. Queued tasks observe CLOSED and do not start
another read.
+ _prefetched.wait(lck, [this]() { return _buffer_status !=
BufferStatus::PENDING; });
+ _merge_file_cache_stats();
_buffer_status = BufferStatus::CLOSED;
_prefetched.notify_all();
// Explicitly release the backing buffer here, in the calling (query)
thread which has a
@@ -647,14 +664,14 @@
PrefetchBufferedReader::PrefetchBufferedReader(RuntimeProfile* profile, io::File
}
_io_ctx = _io_ctx_holder.get();
if (buffer_size == -1L) {
- buffer_size = config::remote_storage_read_buffer_mb * 1024 * 1024;
+ buffer_size = int64_t {config::remote_storage_read_buffer_mb} * 1024 *
1024;
}
_size = _reader->size();
- _whole_pre_buffer_size = buffer_size;
+ // Use the allocated capacity as the ring modulus. For example, 6 MiB
allocates
+ // one 4 MiB buffer, so offset 4 MiB must wrap to slot 0, not index a
missing slot 1.
+ const int64_t buffer_num = std::max<int64_t>(1, buffer_size /
s_max_pre_buffer_size);
+ _whole_pre_buffer_size = buffer_num * s_max_pre_buffer_size;
_file_range.end_offset = std::min(_file_range.end_offset, _size);
- int buffer_num = buffer_size > s_max_pre_buffer_size
- ? cast_set<int>(buffer_size) /
cast_set<int>(s_max_pre_buffer_size)
- : 1;
std::function<void(PrefetchBuffer&)> sync_buffer = nullptr;
if (profile != nullptr) {
const char* prefetch_buffered_reader = "PrefetchBufferedReader";
diff --git a/be/src/io/fs/buffered_reader.h b/be/src/io/fs/buffered_reader.h
index 9b0e66ebbd0..ebe7b4ebe40 100644
--- a/be/src/io/fs/buffered_reader.h
+++ b/be/src/io/fs/buffered_reader.h
@@ -469,6 +469,11 @@ struct PrefetchBuffer :
std::enable_shared_from_this<PrefetchBuffer>, public Pro
io::FileReaderSPtr _reader;
std::shared_ptr<const IOContext> _io_ctx_holder;
const IOContext* _io_ctx = nullptr;
+ // Only the worker writes these statistics. The consumer merges them after
observing
+ // completion under _lock, before reusing or closing the buffer.
+ IOContext _prefetch_io_ctx;
+ FileCacheStatistics _file_cache_stats;
+ bool _has_unmerged_cache_stats = false;
PODArray<char> _buf;
BufferStatus _buffer_status {BufferStatus::RESET};
std::mutex _lock;
@@ -515,6 +520,9 @@ struct PrefetchBuffer :
std::enable_shared_from_this<PrefetchBuffer>, public Pro
void _collect_profile_at_runtime() override {}
void _collect_profile_before_close() override;
+
+private:
+ void _merge_file_cache_stats();
};
constexpr int64_t s_max_pre_buffer_size = 4 * 1024 * 1024; // 4MB
diff --git a/be/test/format_v2/delimited_text/csv_reader_test.cpp
b/be/test/format_v2/delimited_text/csv_reader_test.cpp
index 32d4255acee..8493b216e06 100644
--- a/be/test/format_v2/delimited_text/csv_reader_test.cpp
+++ b/be/test/format_v2/delimited_text/csv_reader_test.cpp
@@ -20,9 +20,12 @@
#include <gtest/gtest.h>
#include <algorithm>
+#include <atomic>
+#include <cstring>
#include <filesystem>
#include <fstream>
#include <memory>
+#include <thread>
#include "common/config.h"
#include "common/consts.h"
@@ -44,17 +47,95 @@
#include "exprs/vexpr_context.h"
#include "format/csv/csv_reader.h"
#include "format_v2/column_mapper.h"
+#include "io/fs/stream_load_pipe.h"
#include "io/io_common.h"
+#include "runtime/exec_env.h"
#include "runtime/runtime_profile.h"
#include "testutil/desc_tbl_builder.h"
#include "testutil/mock/mock_runtime_state.h"
+#include "testutil/mock/obj_storage_client_test_stub.h"
#include "testutil/scoped_temp_dir.h"
#include "util/debug_points.h"
#include "util/defer_op.h"
+#include "util/s3_util.h"
+#include "util/threadpool.h"
namespace doris::format::csv {
namespace {
+class CsvPrefetchS3Client : public io::ObjStorageClientTestStub {
+public:
+ const std::string data =
"id,name,score\n1,alice,10\n2,bob,20\n3,carol,30\n";
+ const std::thread::id scanner_thread = std::this_thread::get_id();
+ std::atomic<size_t> foreground_reads {0};
+ std::atomic<size_t> reads {0};
+
+ io::ObjStorageHeadResult head_object(const io::ObjStoragePath&) override {
+ return {.resp = io::ObjStorageResponse::OK(),
+ .file_size = static_cast<long long>(data.size())};
+ }
+
+ io::ObjStorageResponse get_object(const io::ObjStoragePath&, void* buffer,
size_t offset,
+ size_t bytes_read, size_t* size_return)
override {
+ ++reads;
+ if (std::this_thread::get_id() == scanner_thread) {
+ ++foreground_reads;
+ }
+ EXPECT_LE(offset + bytes_read, data.size());
+ *size_return = std::min(bytes_read, data.size() - offset);
+ std::memcpy(buffer, data.data() + offset, *size_return);
+ return io::ObjStorageResponse::OK();
+ }
+
+ // Only HEAD and GET belong to this read test. Explicit overrides avoid
generating unused
+ // gMock expectation helpers for this anonymous-namespace client.
+ io::ObjStorageUploadResult create_multipart_upload(const
io::ObjStoragePath&) override {
+ return {.resp = unexpected_operation()};
+ }
+
+ io::ObjStorageResponse put_object(const io::ObjStoragePath&,
std::string_view) override {
+ return unexpected_operation();
+ }
+
+ io::ObjStorageUploadResult upload_part(const io::ObjStoragePath&, const
std::string&,
+ std::string_view, int) override {
+ return {.resp = unexpected_operation()};
+ }
+
+ io::ObjStorageResponse complete_multipart_upload(
+ const io::ObjStoragePath&, const std::string&,
+ const std::vector<io::ObjStorageCompletedPart>&) override {
+ return unexpected_operation();
+ }
+
+ io::ObjStorageResponse delete_objects(const io::ObjStoragePath&,
+ std::vector<std::string>) override {
+ return unexpected_operation();
+ }
+
+ io::ObjStorageResponse delete_object(const io::ObjStoragePath&) override {
+ return unexpected_operation();
+ }
+
+ std::string generate_presigned_url(const io::ObjStoragePath&, int64_t)
override {
+ ADD_FAILURE() << "Unexpected signed URL request in CSV prefetch test";
+ return {};
+ }
+
+ io::ObjStorageListPageResult list_objects_page(const io::ObjStoragePath&,
+ std::string_view) override {
+ return {.resp = unexpected_operation()};
+ }
+
+private:
+ static io::ObjStorageResponse unexpected_operation() {
+ ADD_FAILURE() << "Unexpected object storage operation in CSV prefetch
test";
+ return {.status = {TStatusCode::NOT_IMPLEMENTED_ERROR,
+ "CSV prefetch test only supports HEAD and GET"},
+ .http_code = 0};
+ }
+};
+
TFileScanRangeParams csv_scan_params() {
TFileScanRangeParams params;
params.__set_format_type(TFileFormatType::FORMAT_CSV_PLAIN);
@@ -512,6 +593,78 @@ protected:
TFileScanRangeParams _params;
};
+TEST_F(CsvV2ReaderTest, S3PrefetchPreservesSplitRowsAndReadStatistics) {
+ auto* env = ExecEnv::GetInstance();
+ auto saved_pool = std::move(env->_buffered_reader_prefetch_thread_pool);
+ const auto saved_in_memory_size = config::in_memory_file_size;
+ Defer restore([&] {
+ env->_buffered_reader_prefetch_thread_pool.reset();
+ env->_buffered_reader_prefetch_thread_pool = std::move(saved_pool);
+ config::in_memory_file_size = saved_in_memory_size;
+ S3ClientFactory::instance().clear_client_creator_for_test();
+ });
+ ASSERT_TRUE(ThreadPoolBuilder("CsvPrefetchTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&env->_buffered_reader_prefetch_thread_pool)
+ .ok());
+ // Keep the small fixture on the same prefetch path as a large S3 CSV file.
+ config::in_memory_file_size = 0;
+ auto client = std::make_shared<CsvPrefetchS3Client>();
+ S3ClientFactory::instance().set_client_creator_for_test(
+ [client](const S3ClientConf&) { return client; });
+ _params.__set_file_type(TFileType::FILE_S3);
+ _state._query_options.__set_enable_file_cache(false);
+
+ // Cover a complete file, a split starting inside Alice's row, and an
unknown-size first split.
+ for (const int64_t split_start : {0, 17, -1}) {
+ SCOPED_TRACE(split_start);
+ auto properties = std::make_shared<io::FileSystemProperties>();
+ properties->system_type = TFileType::FILE_S3;
+ properties->properties = {{"AWS_ENDPOINT",
"http://csv-prefetch.invalid"},
+ {"AWS_REGION", "test-region"}};
+ auto desc = std::make_unique<io::FileDescription>();
+ desc->path = "s3://csv-prefetch/reader.csv";
+ desc->range_start_offset = std::max<int64_t>(0, split_start);
+ desc->file_size = split_start < 0 ? -1 :
static_cast<int64_t>(client->data.size());
+ desc->range_size = desc->file_size - desc->range_start_offset;
+ io::FileReaderStats stats;
+ auto io_ctx = std::make_shared<io::IOContext>();
+ io_ctx->file_reader_stats = &stats;
+ auto reader =
+ std::make_unique<CsvReader>(properties, desc, io_ctx,
&_profile, &_params, _slots);
+ ASSERT_TRUE(reader->init(&_state).ok());
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<FileScanRequest>();
+ for (int32_t id = 0; id < 3; ++id) {
+ request->non_predicate_columns.push_back(
+ LocalColumnIndex::top_level(LocalColumnId(id)));
+ request->local_positions.emplace(LocalColumnId(id),
LocalIndex(id));
+ }
+ ASSERT_TRUE(reader->open(request).ok());
+ auto block = make_block(schema, {0, 1, 2});
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, split_start > 0 ? 2 : 3);
+ EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, 0),
split_start > 0 ? 2 : 1);
+ EXPECT_EQ(nullable_string_at(*block.get_by_position(1).column, rows -
1), "carol");
+ EXPECT_GT(stats.read_calls, 0);
+ EXPECT_GT(stats.read_bytes, 0);
+ const auto previous_prefetch_io = counter_value(&_profile,
"PreRequestIO");
+ ASSERT_TRUE(reader->close().ok());
+ EXPECT_GT(counter_value(&_profile, "PreRequestIO"),
previous_prefetch_io);
+ EXPECT_GT(counter_value(&_profile, "PreRequestBytes"), 0);
+ EXPECT_GT(counter_value(&_profile, "ReadTime"), 0);
+ const auto collected_prefetch_io = counter_value(&_profile,
"PreRequestIO");
+ ASSERT_TRUE(reader->close().ok());
+ EXPECT_EQ(counter_value(&_profile, "PreRequestIO"),
collected_prefetch_io);
+ }
+ EXPECT_GT(client->reads.load(), 0);
+ EXPECT_EQ(client->foreground_reads.load(), 0);
+}
+
// Scenario: CSV v2 exposes FE-provided file slots as nullable file-local
schema using column_idxs
// as CSV field ordinals.
TEST_F(CsvV2ReaderTest, SchemaUsesSlotTypesAndColumnIdxs) {
@@ -927,6 +1080,17 @@ TEST_F(CsvV2ReaderTest,
StreamInputRequiresLoadIdBeforeOpeningPipe) {
<< status;
}
+TEST_F(CsvV2ReaderTest, ClosePreservesSharedStreamPipe) {
+ _params.__set_file_type(TFileType::FILE_STREAM);
+ auto reader = create_unknown_size_reader(_file_path, &_params, _slots,
&_state, &_profile);
+ auto pipe = std::make_shared<io::StreamLoadPipe>();
+ reader->_file_reader = pipe;
+ ASSERT_TRUE(reader->close().ok());
+ EXPECT_FALSE(pipe->closed());
+ reader.reset();
+ EXPECT_FALSE(pipe->closed());
+}
+
// Scenario: CSV has no footer row count, so v2 COUNT pushdown scans the split
and returns the
// counted row count through FileAggregateResult.
TEST_F(CsvV2ReaderTest, CountAggregateScansRows) {
diff --git a/be/test/io/fs/buffered_reader_test.cpp
b/be/test/io/fs/buffered_reader_test.cpp
index 75ed5e351dc..b8f96b339fc 100644
--- a/be/test/io/fs/buffered_reader_test.cpp
+++ b/be/test/io/fs/buffered_reader_test.cpp
@@ -20,9 +20,12 @@
#include <glog/logging.h>
#include <gtest/gtest-message.h>
#include <gtest/gtest-test-part.h>
+#include <gtest/gtest.h>
#include <limits.h>
+#include <algorithm>
#include <atomic>
+#include <cstring>
#include <memory>
#include <ostream>
#include <thread>
@@ -33,6 +36,7 @@
#include "io/fs/local_file_system.h"
#include "runtime/exec_env.h"
#include "util/countdown_latch.h"
+#include "util/defer_op.h"
#include "util/stopwatch.hpp"
#include "util/threadpool.h"
@@ -150,7 +154,7 @@ public:
protected:
Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
- const io::IOContext* /*io_ctx*/) override {
+ const io::IOContext* io_ctx) override {
_read_started->count_down();
_continue_read->wait();
if (offset >= _size) {
@@ -158,7 +162,9 @@ protected:
return Status::OK();
}
*bytes_read = std::min(_size - offset, result.size);
- return Status::TimedOut("injected prefetch timeout");
+ memset(result.data, 0, *bytes_read);
+ io_ctx->file_cache_stats->bytes_read_from_remote += *bytes_read;
+ return Status::OK();
}
private:
@@ -352,37 +358,160 @@ TEST_F(BufferedReaderTest, test_miss) {
EXPECT_EQ(45, bytes_read);
}
-TEST_F(BufferedReaderTest,
prefetch_task_keeps_reader_alive_after_close_timeout) {
+// A read timeout must not let close/destruction release the scanner's
borrowed state.
+TEST_F(BufferedReaderTest, prefetch_close_drains_timed_out_reads) {
const auto saved_timeout_ms = config::buffered_reader_read_timeout_ms;
- config::buffered_reader_read_timeout_ms = 50;
-
- CountDownLatch read_started(1);
- CountDownLatch continue_read(1);
- std::atomic<bool> reader_destroyed = false;
- auto inner_reader = std::make_shared<BlockingFileReader>(1024,
&read_started, &continue_read,
-
&reader_destroyed);
- auto weak_reader = std::weak_ptr<io::FileReader>(inner_reader);
- {
- io::PrefetchBufferedReader reader(nullptr, std::move(inner_reader),
- io::PrefetchRange(0, 1024));
+ Defer restore([&] { config::buffered_reader_read_timeout_ms =
saved_timeout_ms; });
+ config::buffered_reader_read_timeout_ms = 20;
+
+ for (bool explicit_close : {false, true}) {
+ SCOPED_TRACE(explicit_close);
+ CountDownLatch read_started(1);
+ CountDownLatch continue_read(1);
+ CountDownLatch close_started(1);
+ CountDownLatch close_finished(1);
+ std::atomic<bool> reader_destroyed = false;
+ io::FileCacheStatistics stats;
+ auto io_ctx = std::make_shared<io::IOContext>();
+ io_ctx->file_cache_stats = &stats;
+ auto inner_reader = std::make_shared<BlockingFileReader>(1024,
&read_started,
+
&continue_read, &reader_destroyed);
+ auto reader = std::make_unique<io::PrefetchBufferedReader>(
+ nullptr, std::move(inner_reader), io::PrefetchRange(0, 1024),
io_ctx);
uint8_t buf[1];
size_t bytes_read = 0;
- Status st = reader.read_at(0, Slice {buf, 1}, &bytes_read);
- EXPECT_TRUE(st.is<ErrorCode::TIMEOUT>());
- EXPECT_TRUE(read_started.wait_for(std::chrono::seconds(1)));
- EXPECT_FALSE(reader_destroyed.load());
+ EXPECT_TRUE(reader->read_at(0, Slice {buf, 1},
&bytes_read).is<ErrorCode::TIMEOUT>());
+ EXPECT_TRUE(read_started.wait_for(std::chrono::seconds(5)));
+ EXPECT_EQ(stats.bytes_read_from_remote, 0);
+
+ std::thread finish_read([&] {
+ close_started.wait();
+ // Hold the request beyond the old close timeout. Closing must
remain blocked.
+
EXPECT_FALSE(close_finished.wait_for(std::chrono::milliseconds(100)));
+ EXPECT_FALSE(reader_destroyed.load());
+ continue_read.count_down();
+ });
+ close_started.count_down();
+ if (explicit_close) {
+ EXPECT_TRUE(reader->close().ok());
+ }
+ reader.reset();
+ close_finished.count_down();
+ finish_read.join();
+ // Let the pool release its task captures before the
destructor-observer goes out of scope.
+ ExecEnv::GetInstance()->buffered_reader_prefetch_thread_pool()->wait();
+ EXPECT_TRUE(reader_destroyed.load());
+ EXPECT_EQ(stats.bytes_read_from_remote, 1024);
}
- EXPECT_FALSE(reader_destroyed.load());
- EXPECT_FALSE(weak_reader.expired());
+}
- continue_read.count_down();
- for (int i = 0; i < 100 && !reader_destroyed.load(); ++i) {
- std::this_thread::sleep_for(std::chrono::milliseconds(10));
+class PrefetchStatisticsFileReader : public MockOffsetFileReader {
+public:
+ PrefetchStatisticsFileReader(size_t size, io::FileCacheStatistics*
scanner_stats,
+ CountDownLatch* started, CountDownLatch*
resume)
+ : MockOffsetFileReader(size),
+ _scanner_stats(scanner_stats),
+ _started(started),
+ _resume(resume) {}
+
+ std::atomic<bool> used_scanner_stats = false;
+
+protected:
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const io::IOContext* io_ctx) override {
+ if (io_ctx->file_cache_stats == _scanner_stats) {
+ used_scanner_stats = true;
+ }
+ RETURN_IF_ERROR(MockOffsetFileReader::read_at_impl(offset, result,
bytes_read, io_ctx));
+ auto* stats = io_ctx->file_cache_stats;
+ stats->bytes_read_from_remote += *bytes_read;
+ ++stats->num_remote_io_total;
+ stats->peer_hosts.insert("prefetch-test");
+ stats->remote_only_on_miss_triggered = true;
+ stats->remote_only_on_miss_threshold_bytes = 42;
+ _started->count_down();
+ _resume->wait();
+ return Status::OK();
}
- EXPECT_TRUE(reader_destroyed.load());
- EXPECT_TRUE(weak_reader.expired());
- config::buffered_reader_read_timeout_ms = saved_timeout_ms;
+private:
+ io::FileCacheStatistics* _scanner_stats;
+ CountDownLatch* _started;
+ CountDownLatch* _resume;
+};
+
+TEST_F(BufferedReaderTest,
prefetch_merges_worker_cache_statistics_on_consumer) {
+ constexpr size_t buffer_size = 4 * 1024 * 1024;
+ constexpr size_t file_size = 5 * buffer_size;
+ CountDownLatch started(4);
+ CountDownLatch resume(1);
+ io::FileCacheStatistics stats;
+ auto io_ctx = std::make_shared<io::IOContext>();
+ io_ctx->file_cache_stats = &stats;
+ auto inner =
+ std::make_shared<PrefetchStatisticsFileReader>(file_size, &stats,
&started, &resume);
+ io::PrefetchBufferedReader reader(nullptr, inner, io::PrefetchRange(0,
file_size), io_ctx,
+ 4 * buffer_size);
+ std::thread release_workers([&] {
+ EXPECT_TRUE(started.wait_for(std::chrono::seconds(5)));
+ // All four workers have updated their counters, but the consumer has
not merged them.
+ EXPECT_FALSE(inner->used_scanner_stats.load());
+ EXPECT_EQ(stats.bytes_read_from_remote, 0);
+ resume.count_down();
+ });
+ uint8_t buf[2];
+ size_t bytes_read = 0;
+ auto status = reader.read_at(0, Slice {buf, 1}, &bytes_read);
+ release_workers.join();
+ ASSERT_TRUE(status.ok());
+ EXPECT_EQ(stats.bytes_read_from_remote, buffer_size);
+ ASSERT_TRUE(reader.read_at(1, Slice {buf, 1}, &bytes_read).ok());
+ EXPECT_EQ(stats.bytes_read_from_remote, buffer_size); // No double merge
for another slice.
+ // Reuse slot 0 without consuming slots 1..3; close must still collect
their statistics.
+ ASSERT_TRUE(reader.read_at(4 * buffer_size, Slice {buf, 1},
&bytes_read).ok());
+ EXPECT_EQ(stats.bytes_read_from_remote, 2 * buffer_size);
+ ASSERT_TRUE(reader.close().ok());
+ EXPECT_EQ(stats.bytes_read_from_remote, file_size);
+ EXPECT_EQ(stats.num_remote_io_total, 5);
+ EXPECT_EQ(stats.peer_hosts.count("prefetch-test"), 1);
+ EXPECT_TRUE(stats.remote_only_on_miss_triggered);
+ EXPECT_EQ(stats.remote_only_on_miss_threshold_bytes, 42);
+ ASSERT_TRUE(reader.close().ok());
+ EXPECT_EQ(stats.bytes_read_from_remote, file_size);
+}
+
+TEST_F(BufferedReaderTest, prefetch_normalizes_ring_capacity) {
+ constexpr size_t mb = 1024 * 1024;
+ constexpr size_t file_size = 10 * mb;
+ std::vector<uint8_t> expected(file_size);
+ for (size_t i = 0; i < file_size; ++i) {
+ expected[i] = i % UCHAR_MAX;
+ }
+ const auto saved_buffer_mb = config::remote_storage_read_buffer_mb;
+ Defer restore([&] { config::remote_storage_read_buffer_mb =
saved_buffer_mb; });
+ // Includes less than one buffer, a non-multiple of 4 MiB, and a
multi-slot ring.
+ for (int capacity_mb : {0, 1, 6, 10, 16}) {
+ config::remote_storage_read_buffer_mb = capacity_mb;
+ for (size_t start : {size_t {0}, 4 * mb + 7}) {
+ SCOPED_TRACE(capacity_mb);
+ SCOPED_TRACE(start);
+ io::PrefetchBufferedReader reader(nullptr,
+
std::make_shared<MockOffsetFileReader>(file_size),
+ io::PrefetchRange(start,
file_size));
+ std::vector<uint8_t> data(3 * mb);
+ for (size_t offset = start; offset < file_size;) {
+ size_t bytes_read = 0;
+ ASSERT_TRUE(
+ reader.read_at(offset, Slice {data.data(),
data.size()}, &bytes_read).ok());
+ ASSERT_EQ(bytes_read, std::min(data.size(), file_size -
offset));
+ ASSERT_TRUE(std::equal(data.begin(), data.begin() + bytes_read,
+ expected.begin() + offset))
+ << offset;
+ offset += bytes_read;
+ }
+ ASSERT_TRUE(reader.close().ok());
+ }
+ }
}
TEST_F(BufferedReaderTest, test_read_amplify) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
