This is an automated email from the ASF dual-hosted git repository.
airborne12 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 62cf8c75880 [improvement](be) Read SNII phrase-prefix batches on the
calling thread when they hit the page cache (#69882)
62cf8c75880 is described below
commit 62cf8c75880d6ea02457eab651293a8b3ca2d20c
Author: Jack <[email protected]>
AuthorDate: Mon Nov 2 14:05:38 2026 +0800
[improvement](be) Read SNII phrase-prefix batches on the calling thread
when they hit the page cache (#69882)
### What problem does this PR solve?
Issue Number: None
Related PR: #69648
Problem Summary: #69648 reads a SNII batch on the calling thread when
the index file is local, the
batch has at most `snii_inline_read_max_segments` (4) physical segments
and 256 KiB, and the
moving average of recent reads is under 20 us, i.e. they hit the page
cache. A phrase prefix reads
the postings of every term its prefix expands to, in batches of 5 to 16
segments that stay well
under 256 KiB, so those still went segment by segment through the shared
buffered-reader prefetch
pool: on a hot 99M-row table, about a fifth of the SNII time of such a
query was spent on the
pool's lock and latch (`ThreadPool::do_submit`, `CountDownLatch::wait`
in `read_batch`).
The default of `snii_inline_read_max_segments` becomes 16, which covers
those batches, and two
changes keep larger inline batches safe when the data is not in the page
cache:
- A batch read inline hands the segments it has left to the pool as soon
as one of its reads is
slow. When the average is low but this batch's data is not cached, the
batch pays one read on
the calling thread and overlaps the rest. A single segment left is still
read inline.
- A read counts by how much longer it took than copying its bytes out of
the page cache (at 4
bytes/ns, the low end for one thread), for both the average and the slow
test. A page-cache hit
of 64-256 KiB spends tens of microseconds on the copy alone, which
pushed the average over 20 us
and sent about 5% of a phrase query's small batches to the pool.
On the 99M-row table, the phrase prefix's batches of 5-16 segments are
now all read inline when
hot, and the small batches of a phrase query that still went to the pool
dropped by 77%. In a cold
run (BE restarted, index files evicted from the page cache), the average
was at least 40 us
whenever such a batch was decided, so they all went to the pool, as
before.
| Query (99M rows) | Hot, new/previous rule | Cold, new/previous rule |
|---|---|---|
| phrase prefix 'retry budget dep' | 0.79 / 0.85 / 0.96 (three runs) |
0.96 |
| other phrase queries | 0.87-0.95 | 1.01 (A6) |
| the other 19 queries | 0.98-1.03 | |
Hot: both rules interleaved in one BE process. Cold: BE restarted and
the index files evicted
before every run, configuration order alternated, 5 runs each.
---
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 7523903515a..233db2cf991 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1439,10 +1439,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 36a009c85d4..e6ca4ff9b42 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]
