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 532f60d855b branch-4.2: [improvement](be) Read SNII phrase-prefix
batches on the calling thread when they hit the page cache #69882 (#69962)
532f60d855b is described below
commit 532f60d855b7d778589b31309c205e334c686efe
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Nov 3 10:05:13 2026 +0800
branch-4.2: [improvement](be) Read SNII phrase-prefix batches on the
calling thread when they hit the page cache #69882 (#69962)
Cherry-picked from #69882
Co-authored-by: Jack <[email protected]>
---
be/src/common/config.cpp | 9 +--
.../index/inverted/snii/snii_doris_adapter.cpp | 83 +++++++++++++++-------
.../index/inverted/snii/snii_doris_adapter.h | 3 +
.../inverted/snii/snii_doris_adapter_test.cpp | 51 ++++++++++++-
4 files changed, 114 insertions(+), 32 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index e483782c4b3..12cefd73884 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1451,10 +1451,11 @@ DEFINE_mInt32(snii_prx_zstd_level, "3");
DEFINE_mInt32(snii_prx_zstd_min_saving_percent, "-1");
// 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");
+// Handing 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. 16 covers
the batches of a
+// phrase prefix, which reads the postings of every term the prefix expands to.
+DEFINE_mInt32(snii_inline_read_max_segments, "16");
// 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/storage/index/inverted/snii/snii_doris_adapter.cpp
b/be/src/storage/index/inverted/snii/snii_doris_adapter.cpp
index a7c9e0d228d..cf85372843b 100644
--- a/be/src/storage/index/inverted/snii/snii_doris_adapter.cpp
+++ b/be/src/storage/index/inverted/snii/snii_doris_adapter.cpp
@@ -44,18 +44,26 @@ 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.
+// than the reads when they hit it. A read counts by how much longer it took
than copying its bytes
+// out of the page cache: a hit takes a few microseconds more, a miss on flash
ten times that or
+// more. The average of that excess (1/8 weight) is shared by every reader. 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, and a batch read inline
sends the segments it
+// has left to the pool as soon as one of its reads is slow.
constexpr size_t kInlineMaxBatchBytes = 256 * 1024;
constexpr int64_t kSlowSmallReadNs = 20 * 1000;
+// The low end of what one thread copies out of the page cache, so a hit is
not taken for a miss.
+constexpr int64_t kPageCacheBytesPerNs = 4;
std::atomic<int64_t> g_small_read_ns {0};
+std::atomic<bool> g_slow_read_hand_off {true}; // only tests turn this off
-void note_small_read(int64_t ns) {
+// Notes how much longer than copying its `len` bytes a read took; returns
whether it was slow.
+bool note_read(int64_t ns, size_t len) {
+ const int64_t excess = ns - static_cast<int64_t>(len) /
kPageCacheBytesPerNs;
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);
+ g_small_read_ns.store(average + (excess - average) / 8,
std::memory_order_relaxed);
+ return excess >= kSlowSmallReadNs;
}
} // namespace
@@ -63,6 +71,10 @@ void DorisSniiFileReader::reset_read_latency_for_test() {
g_small_read_ns.store(0, std::memory_order_relaxed);
}
+void DorisSniiFileReader::set_slow_read_hand_off_for_test(bool enabled) {
+ g_slow_read_hand_off.store(enabled, std::memory_order_relaxed);
+}
+
thread_local const io::IOContext* DorisSniiFileReader::_scoped_io_ctx =
nullptr;
ThreadPool* DorisSniiFileReader::_io_pool_for_test = nullptr;
@@ -271,7 +283,8 @@ Status DorisSniiFileReader::read_batch(const
std::vector<::doris::snii::io::Rang
read_bytes += cast_set<int64_t>(seg.len);
}
- // Run the planned physical reads, concurrently when a pool is available.
+ // Run the planned physical reads, concurrently when a pool is available.
Returns whether the
+ // read was timed and slow, i.e. probably missed the page cache.
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.
@@ -279,25 +292,17 @@ Status DorisSniiFileReader::read_batch(const
std::vector<::doris::snii::io::Rang
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());
+ if (!timed || !seg_status[s].ok()) {
+ return false;
}
+ return note_read(std::chrono::duration_cast<std::chrono::nanoseconds>(
+ std::chrono::steady_clock::now() - begin)
+ .count(),
+ segs[s].len);
};
ThreadPool* pool = _select_io_pool();
- // 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) {
+ // Reads segments [first, num_segs) on the pool, at most
kMaxConcurrentReads at a time.
+ auto fan_out = [&](size_t first) {
// 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 ->
@@ -307,7 +312,7 @@ Status DorisSniiFileReader::read_batch(const
std::vector<::doris::snii::io::Rang
// Same shape as the hedged-read path in
cached_remote_file_reader.cpp:500-505.
const std::shared_ptr<ResourceContext> parent_resource_ctx =
thread_context()->resource_ctx();
- for (size_t base = 0; base < num_segs; base += kMaxConcurrentReads) {
+ for (size_t base = first; base < num_segs; base +=
kMaxConcurrentReads) {
const size_t wave_end = std::min(base + kMaxConcurrentReads,
num_segs);
::doris::CountDownLatch latch(cast_set<int>(wave_end - base));
for (size_t s = base; s < wave_end; ++s) {
@@ -329,6 +334,34 @@ Status DorisSniiFileReader::read_batch(const
std::vector<::doris::snii::io::Rang
}
latch.wait();
}
+ };
+ // 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;
+ size_t serial_rounds = _compute_num_waves(num_segs);
+ if (inline_read) {
+ // A read on the calling thread is one dependent round per segment. A
slow one means the
+ // page cache missed although recent reads hit it, so the segments
left overlap in the pool.
+ size_t next = 0;
+ while (next < num_segs) {
+ const bool slow = run_segment(next++);
+ if (slow && pool != nullptr && num_segs - next > 1 &&
+ g_slow_read_hand_off.load(std::memory_order_relaxed)) {
+ break;
+ }
+ }
+ serial_rounds = next + _compute_num_waves(num_segs - next);
+ if (next < num_segs) {
+ fan_out(next);
+ }
+ } else if (pool != nullptr && num_segs > 1) {
+ fan_out(0);
} else {
// Serial fallback: no executor (e.g. tools without ExecEnv) or a
single
// segment (avoids micro-batch scheduling overhead).
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 52c80ad4810..c9b6d1359e9 100644
--- a/be/src/storage/index/inverted/snii/snii_doris_adapter.h
+++ b/be/src/storage/index/inverted/snii/snii_doris_adapter.h
@@ -89,6 +89,9 @@ public:
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();
+ // Test-only: whether a slow read sends the rest of an inline batch to the
pool. Any read can be
+ // slow on a busy host, so a test that a batch stays on the calling thread
turns this off.
+ static void set_slow_read_hand_off_for_test(bool enabled);
private:
static io::IOContext _make_index_io_context(const io::IOContext* io_ctx);
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 451a32c1864..1731059d379 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
@@ -1017,6 +1017,15 @@ private:
int32_t _saved;
};
+// For a test that a batch stays on the calling thread: on a busy host even a
read from memory can
+// take longer than the slow-read threshold, which would send the rest of the
batch to the pool.
+struct NoSlowReadHandOffGuard {
+ NoSlowReadHandOffGuard() {
DorisSniiFileReader::set_slow_read_hand_off_for_test(false); }
+ ~NoSlowReadHandOffGuard() {
DorisSniiFileReader::set_slow_read_hand_off_for_test(true); }
+ NoSlowReadHandOffGuard(const NoSlowReadHandOffGuard&) = delete;
+ NoSlowReadHandOffGuard& operator=(const NoSlowReadHandOffGuard&) = delete;
+};
+
// 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(
@@ -1062,6 +1071,7 @@ size_t reads_on_calling_thread(
// pool costs more than the reads. The serial rounds then equal the segment
count.
TEST(DorisSniiFileReaderInlineReadTest, SmallLocalBatchRunsOnCallingThread) {
InlineReadConfigGuard config_guard(4);
+ NoSlowReadHandOffGuard no_hand_off;
io::FileCacheStatistics stats;
EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 4, &stats), 4U);
EXPECT_EQ(stats.inverted_index_range_read_count, 4);
@@ -1086,8 +1096,8 @@ TEST(DorisSniiFileReaderInlineReadTest,
RemoteSmallBatchStillFansOut) {
}
// 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.
+// fans out again: the first batch starts inline, while the average is still
low, but sends the
+// segments it has left to the pool at its first slow read; the later ones fan
out from the start.
TEST(DorisSniiFileReaderInlineReadTest, SlowReadsReturnToThePool) {
InlineReadConfigGuard config_guard(4);
size_t on_caller_first = 0;
@@ -1101,14 +1111,49 @@ TEST(DorisSniiFileReaderInlineReadTest,
SlowReadsReturnToThePool) {
}
on_caller_last = on_caller;
}
- EXPECT_EQ(on_caller_first, 4U);
+ EXPECT_EQ(on_caller_first, 1U);
EXPECT_EQ(on_caller_last, 0U);
}
+// A phrase prefix reads the postings of every term its prefix expands to, in
batches of up to 16
+// segments; while reads are fast those stay on the calling thread too. One
more fans out.
+TEST(DorisSniiFileReaderInlineReadTest,
PhrasePrefixSizedBatchRunsOnCallingThread) {
+ InlineReadConfigGuard config_guard(16);
+ NoSlowReadHandOffGuard no_hand_off;
+ io::FileCacheStatistics stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 16, &stats), 16U);
+ EXPECT_EQ(stats.inverted_index_serial_read_rounds, 16);
+ io::FileCacheStatistics over_stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 17, &over_stats), 0U);
+ EXPECT_EQ(over_stats.inverted_index_serial_read_rounds, 2);
+}
+
+// The average can be low while this batch's data is not in the page cache.
Its first slow read then
+// sends the segments it has left to the pool, so the batch pays one read on
the calling thread and
+// overlaps the rest: one inline round plus one pool wave.
+TEST(DorisSniiFileReaderInlineReadTest, SlowReadSendsTheSegmentsLeftToThePool)
{
+ InlineReadConfigGuard config_guard(16);
+ io::FileCacheStatistics stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 8, &stats,
std::chrono::microseconds(300)),
+ 1U);
+ EXPECT_EQ(stats.inverted_index_range_read_count, 8);
+ EXPECT_EQ(stats.inverted_index_serial_read_rounds, 2);
+}
+
+// One segment left after a slow read is read on the calling thread: alone it
overlaps nothing.
+TEST(DorisSniiFileReaderInlineReadTest, LastSegmentAfterSlowReadStaysInline) {
+ InlineReadConfigGuard config_guard(16);
+ io::FileCacheStatistics stats;
+ EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/true, 2, &stats,
std::chrono::microseconds(300)),
+ 2U);
+ EXPECT_EQ(stats.inverted_index_serial_read_rounds, 2);
+}
+
// 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);
+ NoSlowReadHandOffGuard no_hand_off;
for (int batch = 0; batch < 12; ++batch) {
io::FileCacheStatistics remote_stats;
EXPECT_EQ(reads_on_calling_thread(/*local_fs=*/false, 4, &remote_stats,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
