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 ed93d882847 [perf](load) Optimize write performance with improved
backpressure control (#66847)
ed93d882847 is described below
commit ed93d882847ec18c3c3a844c1cb7d442b69bdd4c
Author: hui lai <[email protected]>
AuthorDate: Thu Aug 20 14:02:22 2026 +0800
[perf](load) Optimize write performance with improved backpressure control
(#66847)
### What problem does this PR solve?
The previous backpressure control did not accurately represent the real
write pipeline in two cases:
1. A row-binlog `GROUP` request flushes both data and row-binlog
memtables. These two physical tasks are counted independently by
`flush_running_count()`, so comparing the physical count directly with
`memtable_flush_running_count_limit` halves the intended logical flush
concurrency.
2. In storage-compute separation with S3, a memtable flush task can
finish after submitting asynchronous uploads. Therefore, the
flush-running count does not represent downstream S3 pressure. HDFS
writes are synchronous inside the flush task and should continue using
the flush-running count.
This PR improves the backpressure control as follows:
- Storage-compute integrated:
- Use `memtable_flush_running_count_limit * 2` as the effective physical
task limit for row-binlog `GROUP` requests.
- Keep the existing limit unchanged for normal requests.
- Storage-compute separated:
- Detect the underlying filesystem for both normal and group rowset
builders.
- Use the S3 upload queue as the backpressure signal for S3-backed
loads.
- Retain flush-running-count backpressure for HDFS-backed loads,
including the corrected effective limit for row-binlog `GROUP` requests.
- Fall back to flush-running-count backpressure if the S3 upload thread
pool is unavailable.
- Check the S3 queue directly without going through the adaptive
controller state lock.
### Performance test
The baseline used `memtable_flush_running_count_limit = 4`. The new
results are stable Prometheus throughput values from the same three-BE
test cluster.
This benchmark was run in storage-compute-separated mode with S3 as the
remote storage, using a workload where memtable flush itself is not the
bottleneck. The improvement comes from avoiding premature backpressure
and allowing write-side processing to overlap with downstream upload
work. The reported gains should not be generalized to flush-bound
workloads: when flush is already saturated, relaxing backpressure may
mainly increase queued memtables and memory consumption without
producing the same throughput improvement.
| Scenario | Baseline throughput | New throughput | Improvement |
Baseline BE memory | New BE memory |
|---|---:|---:|---:|---:|---:|
| Row binlog off | ~450 MiB/s | ~899 MiB/s | ~100% | ~16 GiB steady, ~19
GiB peak per BE | ~36 GiB steady, ~38.6 GiB peak per BE |
| Row binlog on | ~360 MiB/s | ~844 MiB/s | ~134% | ~19 GiB steady/peak
per BE | ~33 GiB steady, ~36.3 GiB peak per BE |
Stable measurement windows:
- Row binlog on: 17:02:38–17:08:58 CST
- Row binlog off: 17:20:48–17:26:28 CST
The higher concurrency trades memory for throughput. The observed
steady-state cost was approximately +20 GiB per BE with row binlog off
and +14 GiB per BE with row binlog on, or approximately +60 GiB and +42
GiB respectively across the three-BE cluster. `BE Mem` represents total
process allocated memory, not only memtable memory.
HDFS and storage-compute integrated performance were not measured in
this test.
---
be/src/cloud/cloud_delta_writer.cpp | 18 ++++++++++++++++--
be/src/cloud/cloud_rowset_builder.cpp | 8 ++++++++
be/src/cloud/cloud_rowset_builder.h | 4 ++++
be/src/load/delta_writer/delta_writer.cpp | 6 ++++--
4 files changed, 32 insertions(+), 4 deletions(-)
diff --git a/be/src/cloud/cloud_delta_writer.cpp
b/be/src/cloud/cloud_delta_writer.cpp
index 9e7641aca1f..19eaef0f00a 100644
--- a/be/src/cloud/cloud_delta_writer.cpp
+++ b/be/src/cloud/cloud_delta_writer.cpp
@@ -25,6 +25,8 @@
#include "load/memtable/memtable_memory_limiter.h"
#include "runtime/exec_env.h"
#include "runtime/thread_context.h"
+#include "storage/adaptive_thread_pool_controller.h"
+#include "util/threadpool.h"
namespace doris {
@@ -99,8 +101,20 @@ Status CloudDeltaWriter::write(const Block* block, const
TabletAddRowsPayload& r
CHECK(_is_init || _is_cancelled);
{
SCOPED_TIMER(_wait_flush_limit_timer);
- while (_memtable_writer->flush_running_count() >=
- config::memtable_flush_running_count_limit) {
+ auto* s3_file_upload_pool = rowset_builder()->is_s3_storage()
+ ?
ExecEnv::GetInstance()->s3_file_upload_thread_pool()
+ : nullptr;
+ const auto need_backpressure = [this, s3_file_upload_pool] {
+ if (s3_file_upload_pool != nullptr) {
+ return s3_file_upload_pool->get_queue_size() >
+ AdaptiveThreadPoolController::kS3QueueBusyThreshold;
+ }
+ const auto effective_flush_running_count_limit =
+ config::memtable_flush_running_count_limit *
+ (_req.write_req_type == WriteRequestType::GROUP ? 2 : 1);
+ return _memtable_writer->flush_running_count() >=
effective_flush_running_count_limit;
+ };
+ while (need_backpressure()) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
diff --git a/be/src/cloud/cloud_rowset_builder.cpp
b/be/src/cloud/cloud_rowset_builder.cpp
index d36048a6646..0ab26eef6a5 100644
--- a/be/src/cloud/cloud_rowset_builder.cpp
+++ b/be/src/cloud/cloud_rowset_builder.cpp
@@ -23,6 +23,7 @@
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
+#include "io/fs/file_system.h"
#include "storage/rowset/group_rowset_writer.h"
#include "storage/rowset/rowset_factory.h"
#include "storage/rowset/rowset_writer_context.h"
@@ -236,6 +237,13 @@ const RowsetMetaSharedPtr&
CloudRowsetBuilder::rowset_meta() {
return _rowset_writer->rowset_meta();
}
+bool CloudRowsetBuilder::is_s3_storage() const {
+ if (_rowset_writer == nullptr) {
+ return false;
+ }
+ return _rowset_writer->context().fs()->type() == io::FileSystemType::S3;
+}
+
Status CloudRowsetBuilder::commit_rowset(const std::string& job_id, int64_t
table_id) {
return _engine.meta_mgr().commit_rowset(*rowset_meta(), job_id, table_id);
}
diff --git a/be/src/cloud/cloud_rowset_builder.h
b/be/src/cloud/cloud_rowset_builder.h
index b4a9ead311e..c65e1924448 100644
--- a/be/src/cloud/cloud_rowset_builder.h
+++ b/be/src/cloud/cloud_rowset_builder.h
@@ -37,6 +37,8 @@ public:
const RowsetMetaSharedPtr& rowset_meta();
+ virtual bool is_s3_storage() const;
+
virtual Status commit_rowset(const std::string& job_id, int64_t table_id);
virtual Status set_txn_related_info();
@@ -92,6 +94,8 @@ public:
return _data_builder->get_partial_update_info();
}
+ bool is_s3_storage() const override { return
_data_builder->is_s3_storage(); }
+
CloudRowsetBuilder* data_builder() { return _data_builder.get(); }
CloudRowsetBuilder* row_binlog_builder() { return
_row_binlog_builder.get(); }
diff --git a/be/src/load/delta_writer/delta_writer.cpp
b/be/src/load/delta_writer/delta_writer.cpp
index d831df4c8a3..abcda75afa4 100644
--- a/be/src/load/delta_writer/delta_writer.cpp
+++ b/be/src/load/delta_writer/delta_writer.cpp
@@ -178,8 +178,10 @@ Status DeltaWriter::write(const Block* block, const
TabletAddRowsPayload& rows,
}
{
SCOPED_TIMER(_wait_flush_limit_timer);
- while (_memtable_writer->flush_running_count() >=
- config::memtable_flush_running_count_limit) {
+ const auto effective_flush_running_count_limit =
+ config::memtable_flush_running_count_limit *
+ (_req.write_req_type == WriteRequestType::GROUP ? 2 : 1);
+ while (_memtable_writer->flush_running_count() >=
effective_flush_running_count_limit) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
