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 a47d4613d2d [improvement](be) Read small SNII batches on the calling
thread when reads hit the page cache (#69648)
a47d4613d2d is described below
commit a47d4613d2d475efb58a189fd8d43f4431ca642a
Author: Jack <[email protected]>
AuthorDate: Wed Oct 28 18:40:38 2026 +0800
[improvement](be) Read small SNII batches on the calling thread when reads
hit the page cache (#69648)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: DorisSniiFileReader::read_batch submits every physical
segment of a batch with
more than one segment to the shared buffered-reader prefetch thread pool
and waits on a latch.
When the index file is in the page cache the reads take microseconds and
99% of the batches of a
phrase workload are under 256 KB, so the submit, wake-up and wait cost
more than the reads, and a
query that issues thousands of batches from 16 scanner threads queues on
the pool's lock. On 99M
AgentLogs rows, 83% of the threads inside the SNII query path were
blocked in
ThreadPool::do_submit for the date-range phrase query.
A batch of at most config::snii_inline_read_max_segments (default 4, 0
disables) segments and
256 KiB is now read on the calling thread when the index file is on the
local filesystem and the
moving average of recent small physical reads is below 20 us. Remote and
cached-remote readers,
larger batches and slow (page-cache miss) reads still fan out to the
pool, because overlapping
disk reads gain from it: with a fixed rule instead of the average, the
cold run of two phrase
queries was 4-8% slower. The serial read rounds counter counts one round
per segment for a batch
read inline.
Measured on 99M AgentLogs rows (branch-4.2 build; the changed code is
the same on master), same process, 25 queries, inline off against on, 16
samples each:
date-range phrase query 0.39x, any-of-three-phrases top-N 0.70x, no
query above 1.02x. Cold runs
(restart, page cache evicted, three repetitions) of eight queries are
within 2% of the pool.
---
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 b8cf6e3273e..fad0547b0ee 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1434,6 +1434,10 @@ DEFINE_mInt32(snii_prx_zstd_level, "3");
DEFINE_mInt32(snii_prx_zstd_min_saving_percent, "30");
// 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 632cd8fd404..997d7d39fb6 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1473,6 +1473,10 @@ DECLARE_mInt32(snii_prx_zstd_min_saving_percent);
// 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 d819044024c..36a009c85d4 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 ->
@@ -317,8 +362,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) {
@@ -330,7 +374,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]
