This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 763678f7cf7 branch-4.2: [improvement](be) Read small SNII batches on
the calling thread when reads hit the page cache #69648 (#69763)
763678f7cf7 is described below
commit 763678f7cf7f8fdd8e4390471d90bbe2a4fbfebf
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Oct 29 09:44:05 2026 +0800
branch-4.2: [improvement](be) Read small SNII batches on the calling thread
when reads hit the page cache #69648 (#69763)
Cherry-picked from #69648
Co-authored-by: Jack <[email protected]>
---
be/src/common/config.cpp | 4 +
be/src/common/config.h | 4 +
be/src/storage/index/index_file_reader.cpp | 3 +-
.../index/inverted/snii/snii_doris_adapter.cpp | 54 ++++++-
.../index/inverted/snii/snii_doris_adapter.h | 8 +-
.../inverted/snii/snii_doris_adapter_test.cpp | 161 +++++++++++++++++++++
6 files changed, 227 insertions(+), 7 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 98d10247eab..4e50ec73e28 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1443,6 +1443,10 @@ DEFINE_mInt32(snii_dict_block_zstd_level, "3");
DEFINE_mInt32(snii_prx_zstd_level, "3");
// Zstd level for direct-load .prx regions. Compaction rewrites them using
snii_prx_zstd_level.
DEFINE_mInt32(snii_prx_zstd_level_direct_load, "3");
+// Handing a few small page-cache reads to the shared prefetch pool costs more
than the reads: a
+// query that issues thousands of them queues on the pool lock. Remote, larger
and slow (cold)
+// batches still fan out, so overlapping disk reads keep their parallelism.
+DEFINE_mInt32(snii_inline_read_max_segments, "4");
// G16-d: target SNII dict block size in bytes; 0 uses the format default
// (64 KiB). Larger blocks compress better under the per-block zstd (the dict
// is the dominant physical section on high-cardinality corpora) at the cost
diff --git a/be/src/common/config.h b/be/src/common/config.h
index 966b91743a8..35b16fdeea6 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1442,6 +1442,10 @@ DECLARE_mInt32(snii_prx_zstd_level);
// import); compaction rewrites at snii_prx_zstd_level so settled segments are
// unaffected. Full contract at the DEFINE in config.cpp.
DECLARE_mInt32(snii_prx_zstd_level_direct_load);
+// A SNII batch read of at most this many physical segments (and 256 KiB) is
read on the calling
+// thread when the index file is on the local filesystem and recent small
reads were fast, i.e. hit
+// the page cache. 0 always fans out to the buffered-reader prefetch pool.
+DECLARE_mInt32(snii_inline_read_max_segments);
// G16-d: target SNII dict block size in bytes; 0 = format default (64 KiB).
// Bigger blocks -> better per-block zstd on the dict region, larger cold
// fetch+decompress unit per dict-block miss. Write side only.
diff --git a/be/src/storage/index/index_file_reader.cpp
b/be/src/storage/index/index_file_reader.cpp
index ad3f19d3d97..12ddda2711e 100644
--- a/be/src/storage/index/index_file_reader.cpp
+++ b/be/src/storage/index/index_file_reader.cpp
@@ -178,7 +178,8 @@ Status IndexFileReader::_init_snii(const io::IOContext*
io_ctx) {
const bool direct_remote_io = opts.cache_type ==
io::FileCachePolicy::NO_CACHE &&
_fs->type() != io::FileSystemType::LOCAL;
_snii_file_reader = std::make_shared<snii_doris::DorisSniiFileReader>(
- std::move(reader), /*io_ctx=*/nullptr, direct_remote_io);
+ std::move(reader), /*io_ctx=*/nullptr, direct_remote_io,
+ /*local_fs=*/_fs->type() == io::FileSystemType::LOCAL);
_snii_segment_reader =
std::make_unique<doris::snii::reader::SniiSegmentReader>();
io::IOContext meta_io_ctx;
if (io_ctx != nullptr) {
diff --git a/be/src/storage/index/inverted/snii/snii_doris_adapter.cpp
b/be/src/storage/index/inverted/snii/snii_doris_adapter.cpp
index 0405d07007a..a7c9e0d228d 100644
--- a/be/src/storage/index/inverted/snii/snii_doris_adapter.cpp
+++ b/be/src/storage/index/inverted/snii/snii_doris_adapter.cpp
@@ -20,11 +20,14 @@
#include <fmt/format.h>
#include <algorithm>
+#include <atomic>
+#include <chrono>
#include <cstddef>
#include <limits>
#include <vector>
#include "common/cast_set.h"
+#include "common/config.h"
#include "cpp/sync_point.h"
#include "runtime/exec_env.h"
#include "runtime/thread_context.h"
@@ -38,8 +41,28 @@ namespace {
// (mirrors the "at most one serial round" contract that the test-side
// MeteredFileReader measures, and the S3 standalone reader's 16-way fan-out).
constexpr size_t kMaxConcurrentReads = 16;
+
+// A batch is read on the calling thread only if it is small and reads have
been fast lately. The
+// pool pays off when reads miss the page cache, because a few of them then
overlap; it costs more
+// than the reads when they hit it. The average is of the duration of one
small physical read
+// (1/8 weight), shared by every reader: a page-cache hit takes a few
microseconds, a miss on flash
+// ten times that or more. On NVMe, cold phrase queries ran 4-8% slower with
the threshold at 40 us
+// and level with the pool at 25 us and below, while the warm gain held from
15 us up. It starts at
+// zero, so a cold start reads its first batches inline until the average has
caught up.
+constexpr size_t kInlineMaxBatchBytes = 256 * 1024;
+constexpr int64_t kSlowSmallReadNs = 20 * 1000;
+std::atomic<int64_t> g_small_read_ns {0};
+
+void note_small_read(int64_t ns) {
+ const int64_t average = g_small_read_ns.load(std::memory_order_relaxed);
+ g_small_read_ns.store(average + (ns - average) / 8,
std::memory_order_relaxed);
+}
} // namespace
+void DorisSniiFileReader::reset_read_latency_for_test() {
+ g_small_read_ns.store(0, std::memory_order_relaxed);
+}
+
thread_local const io::IOContext* DorisSniiFileReader::_scoped_io_ctx =
nullptr;
ThreadPool* DorisSniiFileReader::_io_pool_for_test = nullptr;
@@ -62,9 +85,10 @@ uint64_t DorisSniiFileWriter::bytes_written() const {
}
DorisSniiFileReader::DorisSniiFileReader(io::FileReaderSPtr reader, const
io::IOContext* io_ctx,
- bool direct_remote_io)
+ bool direct_remote_io, bool local_fs)
: _reader(std::move(reader)),
_direct_remote_io(direct_remote_io),
+ _local_fs(local_fs),
_default_io_ctx(_make_index_io_context(io_ctx)) {}
io::IOContext DorisSniiFileReader::_make_index_io_context(const io::IOContext*
io_ctx) {
@@ -249,10 +273,31 @@ Status DorisSniiFileReader::read_batch(const
std::vector<::doris::snii::io::Rang
// Run the planned physical reads, concurrently when a pool is available.
auto run_segment = [&](size_t s) {
+ // Only local reads are timed: the average decides about local
readers, and a remote or
+ // cached-remote read takes a millisecond that says nothing about the
page cache.
+ const bool timed = _local_fs && segs[s].len <= kInlineMaxBatchBytes;
+ const auto begin =
+ timed ? std::chrono::steady_clock::now() :
std::chrono::steady_clock::time_point {};
seg_status[s] = _read_at(segs[s].offset, segs[s].len,
&outs->buffers[s], &seg_io_ctx[s]);
+ if (timed && seg_status[s].ok()) {
+
note_small_read(std::chrono::duration_cast<std::chrono::nanoseconds>(
+ std::chrono::steady_clock::now() - begin)
+ .count());
+ }
};
ThreadPool* pool = _select_io_pool();
- if (pool != nullptr && num_segs > 1) {
+ // Submitting to the shared pool and waiting costs far more than a small
read from the page
+ // cache, and a query that issues thousands of such batches queues on the
pool's lock. Only
+ // local files take this path: a remote or cached-remote read is slow
enough that fanning out
+ // pays, and so is a local read that misses the page cache.
+ const int32_t inline_max_segments = config::snii_inline_read_max_segments;
+ const bool inline_read = _local_fs && inline_max_segments > 0 &&
+ num_segs <=
static_cast<size_t>(inline_max_segments) &&
+ read_bytes <=
static_cast<int64_t>(kInlineMaxBatchBytes) &&
+ g_small_read_ns.load(std::memory_order_relaxed) <
kSlowSmallReadNs;
+ // A read on the calling thread is one dependent round per segment.
+ const size_t serial_rounds = inline_read ? num_segs :
_compute_num_waves(num_segs);
+ if (pool != nullptr && num_segs > 1 && !inline_read) {
// Carried onto the pool threads below. They are "Orphan" threads with
no MemTrackerLimiter
// of their own (buffered_reader.cpp:426), and the read does allocate
down there: in cloud
// mode _read_at reaches CachedRemoteFileReader::read_at_impl ->
@@ -320,8 +365,7 @@ Status DorisSniiFileReader::read_batch(const
std::vector<::doris::snii::io::Rang
// Record only what actually completed so the logical counters stay
// truthful for the failed batch; ranges/rounds reflect what was
issued.
_record_read_stats(completed_request_bytes, completed_read_bytes,
- cast_set<int64_t>(num_segs),
- cast_set<int64_t>(_compute_num_waves(num_segs)));
+ cast_set<int64_t>(num_segs),
cast_set<int64_t>(serial_rounds));
return first_error;
}
for (size_t s = 0; s < num_segs; ++s) {
@@ -333,7 +377,7 @@ Status DorisSniiFileReader::read_batch(const
std::vector<::doris::snii::io::Rang
}
}
_record_read_stats(request_bytes, read_bytes, cast_set<int64_t>(num_segs),
- cast_set<int64_t>(_compute_num_waves(num_segs)));
+ cast_set<int64_t>(serial_rounds));
return Status::OK();
}
// NOLINTEND(readability-non-const-parameter)
diff --git a/be/src/storage/index/inverted/snii/snii_doris_adapter.h
b/be/src/storage/index/inverted/snii/snii_doris_adapter.h
index bb6f2c13dc6..52c80ad4810 100644
--- a/be/src/storage/index/inverted/snii/snii_doris_adapter.h
+++ b/be/src/storage/index/inverted/snii/snii_doris_adapter.h
@@ -67,8 +67,11 @@ public:
// account physical remote bytes, so this reader then counts its own reads
as
// physical remote IO. Never set it for cached or local readers: the former
// double-counts, the latter reports local disk as remote fetch volume.
+ // `local_fs` marks a reader over the local filesystem. A small batch read
from it runs on the
+ // calling thread while reads are fast
(config::snii_inline_read_max_segments); remote readers
+ // always fan out.
explicit DorisSniiFileReader(io::FileReaderSPtr reader, const
io::IOContext* io_ctx = nullptr,
- bool direct_remote_io = false);
+ bool direct_remote_io = false, bool local_fs
= false);
Status read_at(uint64_t offset, size_t len, std::vector<uint8_t>* out)
override;
// Fills the caller's buffer directly (no vector round-trip) under the same
@@ -84,6 +87,8 @@ public:
// batch segment reads. nullptr (default) routes to the BE buffered-reader
// prefetch pool when ExecEnv is ready, otherwise a serial fallback.
static void set_io_thread_pool_for_test(ThreadPool* pool);
+ // Test-only: forget the measured read latency, so the next batch is
judged afresh.
+ static void reset_read_latency_for_test();
private:
static io::IOContext _make_index_io_context(const io::IOContext* io_ctx);
@@ -110,6 +115,7 @@ private:
io::FileReaderSPtr _reader;
const bool _direct_remote_io = false;
+ const bool _local_fs = false;
io::IOContext _default_io_ctx;
static thread_local const io::IOContext* _scoped_io_ctx;
// Test seam for the batch-read executor; see set_io_thread_pool_for_test.
diff --git a/be/test/storage/index/inverted/snii/snii_doris_adapter_test.cpp
b/be/test/storage/index/inverted/snii/snii_doris_adapter_test.cpp
index 4938ee04d54..451a32c1864 100644
--- a/be/test/storage/index/inverted/snii/snii_doris_adapter_test.cpp
+++ b/be/test/storage/index/inverted/snii/snii_doris_adapter_test.cpp
@@ -34,6 +34,7 @@
#include <utility>
#include <vector>
+#include "common/config.h"
#include "common/status.h"
#include "io/fs/file_reader.h"
#include "io/fs/local_file_system.h"
@@ -100,6 +101,9 @@ public:
void xor_byte(size_t offset, uint8_t mask) { _data[offset] ^=
static_cast<char>(mask); }
+ // Makes every read take at least `delay`, like a page-cache miss.
+ void set_read_delay(std::chrono::microseconds delay) { _read_delay =
delay; }
+
void block_once_on_range_containing(size_t offset) {
std::lock_guard<std::mutex> guard(_control_mu);
_block_offset = offset;
@@ -161,6 +165,9 @@ protected:
}
}
+ if (_read_delay.count() > 0) {
+ std::this_thread::sleep_for(_read_delay);
+ }
if (result.size > 0) {
std::memcpy(result.data, _data.data() + offset, result.size);
}
@@ -204,6 +211,7 @@ private:
mutable std::mutex _control_mu;
std::condition_variable _control_cv;
size_t _block_offset = std::numeric_limits<size_t>::max();
+ std::chrono::microseconds _read_delay {0};
bool _block_armed = false;
bool _blocked = false;
bool _released = false;
@@ -991,4 +999,157 @@ TEST(DorisSniiFileReaderConcurrencyTest,
ParallelPathMatchesSerialPath) {
}
}
+// Sets config::snii_inline_read_max_segments and clears the shared
read-latency average; both are
+// restored on exit so a test never leaks them into its siblings.
+struct InlineReadConfigGuard {
+ explicit InlineReadConfigGuard(int32_t value) :
_saved(config::snii_inline_read_max_segments) {
+ config::snii_inline_read_max_segments = value;
+ DorisSniiFileReader::reset_read_latency_for_test();
+ }
+ ~InlineReadConfigGuard() {
+ config::snii_inline_read_max_segments = _saved;
+ DorisSniiFileReader::reset_read_latency_for_test();
+ }
+ InlineReadConfigGuard(const InlineReadConfigGuard&) = delete;
+ InlineReadConfigGuard& operator=(const InlineReadConfigGuard&) = delete;
+
+private:
+ int32_t _saved;
+};
+
+// Reads `segments` disjoint 4-byte ranges through a reader backed by a
two-thread pool and returns
+// how many of the physical reads ran on the calling thread.
+size_t reads_on_calling_thread(
+ bool local_fs, size_t segments, io::FileCacheStatistics* stats,
+ std::chrono::microseconds read_delay = std::chrono::microseconds(0)) {
+ const std::string data = make_pattern(segments * 8192 + 16);
+ auto recording_reader = std::make_shared<RecordingFileReader>(data);
+ recording_reader->set_read_delay(read_delay);
+ DorisSniiFileReader reader(recording_reader, /*io_ctx=*/nullptr,
/*direct_remote_io=*/false,
+ local_fs);
+
+ std::unique_ptr<ThreadPool> pool;
+ auto pool_st =
+
ThreadPoolBuilder("snii_batch_test").set_min_threads(2).set_max_threads(4).build(&pool);
+ EXPECT_TRUE(pool_st.ok()) << pool_st.to_string();
+ IoPoolSeamGuard seam(pool.get());
+
+ io::IOContext io_ctx;
+ io_ctx.file_cache_stats = stats;
+ std::vector<::doris::snii::io::Range> ranges;
+ for (size_t i = 0; i < segments; ++i) {
+ ranges.push_back({static_cast<uint64_t>(i) * 8192, 4});
+ }
+ index_query::IoReadResult outs;
+ {
+ DorisSniiFileReader::ScopedIOContext scope(&io_ctx);
+ auto status = reader.read_batch(ranges, &outs);
+ EXPECT_TRUE(status.ok()) << status.to_string();
+ }
+ EXPECT_EQ(outs.views.size(), segments);
+ for (size_t i = 0; i < segments && i < outs.views.size(); ++i) {
+ EXPECT_EQ(as_string(outs.views[i]), data.substr(i * 8192, 4));
+ }
+ EXPECT_EQ(recording_reader->reads().size(), segments);
+ size_t on_caller = 0;
+ for (const CapturedRead& read : recording_reader->reads()) {
+ on_caller += read.thread_id == std::this_thread::get_id() ? 1 : 0;
+ }
+ return on_caller;
+}
+
+// A small batch on the local filesystem is read by the calling thread:
handing it to the shared
+// pool costs more than the reads. The serial rounds then equal the segment
count.
+TEST(DorisSniiFileReaderInlineReadTest, SmallLocalBatchRunsOnCallingThread) {
+ InlineReadConfigGuard config_guard(4);
+ io::FileCacheStatistics stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 4, &stats), 4U);
+ EXPECT_EQ(stats.inverted_index_range_read_count, 4);
+ EXPECT_EQ(stats.inverted_index_serial_read_rounds, 4);
+}
+
+// Above the segment limit the batch still fans out to the pool.
+TEST(DorisSniiFileReaderInlineReadTest, LargeLocalBatchStillFansOut) {
+ InlineReadConfigGuard config_guard(4);
+ io::FileCacheStatistics stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 8, &stats), 0U);
+ EXPECT_EQ(stats.inverted_index_range_read_count, 8);
+ EXPECT_EQ(stats.inverted_index_serial_read_rounds, 1);
+}
+
+// A remote or cached-remote reader always fans out, whatever the batch size.
+TEST(DorisSniiFileReaderInlineReadTest, RemoteSmallBatchStillFansOut) {
+ InlineReadConfigGuard config_guard(4);
+ io::FileCacheStatistics stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/false, 4, &stats), 0U);
+ EXPECT_EQ(stats.inverted_index_serial_read_rounds, 1);
+}
+
+// Reads that miss the page cache overlap in the pool, so once recent reads
were slow a small batch
+// fans out again: the first batches are read inline while the average catches
up, the later ones
+// are not.
+TEST(DorisSniiFileReaderInlineReadTest, SlowReadsReturnToThePool) {
+ InlineReadConfigGuard config_guard(4);
+ size_t on_caller_first = 0;
+ size_t on_caller_last = 0;
+ for (int batch = 0; batch < 12; ++batch) {
+ io::FileCacheStatistics stats;
+ const size_t on_caller = reads_on_calling_thread(/*local_fs=*/true, 4,
&stats,
+
std::chrono::microseconds(300));
+ if (batch == 0) {
+ on_caller_first = on_caller;
+ }
+ on_caller_last = on_caller;
+ }
+ EXPECT_EQ(on_caller_first, 4U);
+ EXPECT_EQ(on_caller_last, 0U);
+}
+
+// Slow remote reads must not lower the local reads' chance of being read
inline: with tiered
+// storage one BE serves local and remote rowsets, and only local readers
consult the average.
+TEST(DorisSniiFileReaderInlineReadTest, RemoteReadsDoNotSlowTheLocalEstimate) {
+ InlineReadConfigGuard config_guard(4);
+ for (int batch = 0; batch < 12; ++batch) {
+ io::FileCacheStatistics remote_stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/false, 4, &remote_stats,
+ std::chrono::microseconds(1000)),
+ 0U);
+ }
+ io::FileCacheStatistics local_stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 4, &local_stats), 4U);
+}
+
+// A batch above the byte limit fans out even when reads are fast.
+TEST(DorisSniiFileReaderInlineReadTest, LargeBytesBatchStillFansOut) {
+ InlineReadConfigGuard config_guard(4);
+ const std::string data = make_pattern(2 << 20);
+ auto recording_reader = std::make_shared<RecordingFileReader>(data);
+ DorisSniiFileReader reader(recording_reader, nullptr, false,
/*local_fs=*/true);
+ std::unique_ptr<ThreadPool> pool;
+ ASSERT_TRUE(ThreadPoolBuilder("snii_batch_test")
+ .set_min_threads(2)
+ .set_max_threads(4)
+ .build(&pool)
+ .ok());
+ IoPoolSeamGuard seam(pool.get());
+ // Two disjoint 300 KiB ranges, so they do not coalesce and the batch
exceeds 256 KiB.
+ std::vector<::doris::snii::io::Range> ranges {{0, 300 * 1024}, {1 << 20,
300 * 1024}};
+ index_query::IoReadResult outs;
+ ASSERT_TRUE(reader.read_batch(ranges, &outs).ok());
+ size_t on_caller = 0;
+ for (const CapturedRead& read : recording_reader->reads()) {
+ on_caller += read.thread_id == std::this_thread::get_id() ? 1 : 0;
+ }
+ EXPECT_EQ(recording_reader->reads().size(), 2U);
+ EXPECT_EQ(on_caller, 0U);
+}
+
+// Zero turns the shortcut off and restores the previous behaviour for local
files.
+TEST(DorisSniiFileReaderInlineReadTest, ZeroDisablesInlineRead) {
+ InlineReadConfigGuard config_guard(0);
+ io::FileCacheStatistics stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 4, &stats), 0U);
+ EXPECT_EQ(stats.inverted_index_serial_read_rounds, 1);
+}
+
} // namespace doris::segment_v2::snii_doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
