From 05434d9b497f9e146bbe61ef6e83957ff9e29dbe Mon Sep 17 00:00:00 2001 From: laihui Date: Mon, 17 Aug 2026 18:40:42 +0800 Subject: [PATCH 1/3] [fix](be) Fix row binlog memtable flush backpressure ### What problem does this PR solve? Issue Number: None Related PR: #62007 Problem Summary: A row-binlog group memtable produces two physical flush tasks, but write backpressure used the same physical task limit as a normal data memtable. This effectively halved logical flush concurrency and serialized writes behind flushes. Use twice the configured task limit for group writes in both local and cloud modes. In cloud mode, also apply backpressure when the S3 upload queue is busy, while retaining the flush-running-count memory bound. ### Release note None ### Check List (For Author) - Test: Not run (per request) - Behavior changed: Yes, row-binlog writes use the intended logical flush concurrency and cloud writes also honor S3 upload queue pressure - Does this need documentation: No --- be/src/cloud/cloud_delta_writer.cpp | 8 +++- be/src/load/delta_writer/delta_writer.cpp | 5 +- .../load/delta_writer/delta_writer_context.h | 6 +++ be/src/storage/storage_engine.cpp | 10 ++-- .../load/delta_writer/delta_writer_test.cpp | 11 +++++ .../adaptive_thread_pool_controller_test.cpp | 46 +++++++++++++++++++ 6 files changed, 78 insertions(+), 8 deletions(-) diff --git a/be/src/cloud/cloud_delta_writer.cpp b/be/src/cloud/cloud_delta_writer.cpp index 9e7641aca1fb4f..e81a2f3c5376f9 100644 --- a/be/src/cloud/cloud_delta_writer.cpp +++ b/be/src/cloud/cloud_delta_writer.cpp @@ -99,8 +99,12 @@ 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) { + const auto effective_flush_running_count_limit = + config::memtable_flush_running_count_limit * _req.memtable_flush_task_count(); + auto* adaptive_controller = + ExecEnv::GetInstance()->storage_engine().adaptive_thread_controller(); + while (_memtable_writer->flush_running_count() >= effective_flush_running_count_limit || + adaptive_controller->is_io_busy()) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } diff --git a/be/src/load/delta_writer/delta_writer.cpp b/be/src/load/delta_writer/delta_writer.cpp index d831df4c8a3d90..9dea3d8154fc66 100644 --- a/be/src/load/delta_writer/delta_writer.cpp +++ b/be/src/load/delta_writer/delta_writer.cpp @@ -178,8 +178,9 @@ 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.memtable_flush_task_count(); + while (_memtable_writer->flush_running_count() >= effective_flush_running_count_limit) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } diff --git a/be/src/load/delta_writer/delta_writer_context.h b/be/src/load/delta_writer/delta_writer_context.h index 856601c0be6a83..0a078c9a7420b3 100644 --- a/be/src/load/delta_writer/delta_writer_context.h +++ b/be/src/load/delta_writer/delta_writer_context.h @@ -56,6 +56,12 @@ struct WriteRequest { WriteRequestType write_req_type = WriteRequestType::DATA; std::string storage_vault_id; bool enable_table_memtable_backpressure = false; + + int32_t memtable_flush_task_count() const { + // A GROUP request flushes both the data and row-binlog memtables. The two physical + // flush tasks are counted independently by flush_running_count(). + return write_req_type == WriteRequestType::GROUP ? 2 : 1; + } }; struct TabletAddRowsPayload { diff --git a/be/src/storage/storage_engine.cpp b/be/src/storage/storage_engine.cpp index 2093a17e663ae8..80f92e4b14435f 100644 --- a/be/src/storage/storage_engine.cpp +++ b/be/src/storage/storage_engine.cpp @@ -140,15 +140,17 @@ int64_t BaseStorageEngine::memory_limitation_bytes_per_thread_for_schema_change( } void BaseStorageEngine::_start_adaptive_thread_controller() { - if (!config::enable_adaptive_flush_threads) { - return; - } - auto* system_metrics = DorisMetrics::instance()->system_metrics(); auto* s3_upload_pool = ExecEnv::GetInstance()->s3_file_upload_thread_pool(); _adaptive_thread_controller.init(system_metrics, s3_upload_pool); + // Cloud write backpressure also uses is_io_busy(), so initialize the IO signal even when + // adaptive flush-thread adjustment is disabled. + if (!config::enable_adaptive_flush_threads) { + return; + } + if (_memtable_flush_executor) { auto* flush_pool = _memtable_flush_executor->flush_pool(); auto* high_prio_pool = _memtable_flush_executor->high_prio_flush_pool(); diff --git a/be/test/load/delta_writer/delta_writer_test.cpp b/be/test/load/delta_writer/delta_writer_test.cpp index 62eafce41135f6..a7b75f0bc94019 100644 --- a/be/test/load/delta_writer/delta_writer_test.cpp +++ b/be/test/load/delta_writer/delta_writer_test.cpp @@ -66,6 +66,17 @@ namespace doris { class OlapMeta; +TEST(WriteRequestTest, MemtableFlushTaskCount) { + WriteRequest req; + EXPECT_EQ(req.memtable_flush_task_count(), 1); + + req.write_req_type = WriteRequestType::ROW_BINLOG; + EXPECT_EQ(req.memtable_flush_task_count(), 1); + + req.write_req_type = WriteRequestType::GROUP; + EXPECT_EQ(req.memtable_flush_task_count(), 2); +} + // This is DeltaWriter unit test which used by streaming load. // And also it should take schema change into account after streaming load. diff --git a/be/test/storage/adaptive_thread_pool_controller_test.cpp b/be/test/storage/adaptive_thread_pool_controller_test.cpp index 6c4e42fea05897..38767812d48cda 100644 --- a/be/test/storage/adaptive_thread_pool_controller_test.cpp +++ b/be/test/storage/adaptive_thread_pool_controller_test.cpp @@ -22,10 +22,12 @@ #include #include +#include "cloud/config.h" #include "common/config.h" #include "common/metrics/metrics.h" #include "common/metrics/system_metrics.h" #include "testutil/test_util.h" +#include "util/countdown_latch.h" #include "util/threadpool.h" namespace doris { @@ -44,6 +46,8 @@ class AdaptiveThreadPoolControllerTest : public testing::Test { protected: void SetUp() override { _original_enable_adaptive = config::enable_adaptive_flush_threads; + _original_deploy_mode = config::deploy_mode; + _original_cloud_unique_id = config::cloud_unique_id; int num_cpus = std::thread::hardware_concurrency(); if (num_cpus <= 0) num_cpus = 1; @@ -64,11 +68,15 @@ class AdaptiveThreadPoolControllerTest : public testing::Test { void TearDown() override { config::enable_adaptive_flush_threads = _original_enable_adaptive; + config::deploy_mode = _original_deploy_mode; + config::cloud_unique_id = _original_cloud_unique_id; if (_pool) _pool->shutdown(); if (_pool2) _pool2->shutdown(); } bool _original_enable_adaptive; + std::string _original_deploy_mode; + std::string _original_cloud_unique_id; std::unique_ptr _pool; std::unique_ptr _pool2; }; @@ -260,6 +268,44 @@ TEST_F(AdaptiveThreadPoolControllerTest, TestIoBusyCpuBusyWithNullMetrics) { EXPECT_FALSE(controller.is_cpu_busy()); } +TEST_F(AdaptiveThreadPoolControllerTest, TestCloudIoBusyUsesS3UploadQueue) { + config::deploy_mode = "cloud"; + config::cloud_unique_id.clear(); + + std::unique_ptr upload_pool; + ASSERT_TRUE(ThreadPoolBuilder("TestS3UploadPool") + .set_min_threads(1) + .set_max_threads(1) + .set_max_queue_size(AdaptiveThreadPoolController::kS3QueueBusyThreshold + 1) + .build(&upload_pool) + .ok()); + + CountDownLatch task_started(1); + CountDownLatch release_task(1); + ASSERT_TRUE(upload_pool + ->submit_func([&]() { + task_started.count_down(); + release_task.wait(); + }) + .ok()); + EXPECT_TRUE(task_started.wait_for(std::chrono::seconds(5))); + + for (int i = 0; i < AdaptiveThreadPoolController::kS3QueueBusyThreshold; ++i) { + EXPECT_TRUE(upload_pool->submit_func([]() {}).ok()); + } + + AdaptiveThreadPoolController controller; + controller.init(nullptr, upload_pool.get()); + EXPECT_FALSE(controller.is_io_busy()); + + EXPECT_TRUE(upload_pool->submit_func([]() {}).ok()); + EXPECT_TRUE(controller.is_io_busy()); + + release_task.count_down(); + upload_pool->shutdown(); + upload_pool->wait(); +} + TEST_F(AdaptiveThreadPoolControllerTest, TestCpuBusyUsesCpuMetricsDelta) { MetricRegistry registry("test"); const std::string before_path = get_stat_test_data_path("stat_cpu_busy_before"); From b8df44aaf5fbfe58b00ee8ff4fe488feeddcb4f3 Mon Sep 17 00:00:00 2001 From: laihui Date: Mon, 17 Aug 2026 18:47:54 +0800 Subject: [PATCH 2/3] [fix](be) Remove S3 queue write backpressure Keep this change focused on correcting the physical flush task limit for row-binlog group writes. Remove the additional cloud S3 upload queue condition and its supporting initialization and test changes. --- be/src/cloud/cloud_delta_writer.cpp | 5 +- be/src/storage/storage_engine.cpp | 10 ++-- .../adaptive_thread_pool_controller_test.cpp | 46 ------------------- 3 files changed, 5 insertions(+), 56 deletions(-) diff --git a/be/src/cloud/cloud_delta_writer.cpp b/be/src/cloud/cloud_delta_writer.cpp index e81a2f3c5376f9..7b6e2785fca3d0 100644 --- a/be/src/cloud/cloud_delta_writer.cpp +++ b/be/src/cloud/cloud_delta_writer.cpp @@ -101,10 +101,7 @@ Status CloudDeltaWriter::write(const Block* block, const TabletAddRowsPayload& r SCOPED_TIMER(_wait_flush_limit_timer); const auto effective_flush_running_count_limit = config::memtable_flush_running_count_limit * _req.memtable_flush_task_count(); - auto* adaptive_controller = - ExecEnv::GetInstance()->storage_engine().adaptive_thread_controller(); - while (_memtable_writer->flush_running_count() >= effective_flush_running_count_limit || - adaptive_controller->is_io_busy()) { + while (_memtable_writer->flush_running_count() >= effective_flush_running_count_limit) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } diff --git a/be/src/storage/storage_engine.cpp b/be/src/storage/storage_engine.cpp index 80f92e4b14435f..2093a17e663ae8 100644 --- a/be/src/storage/storage_engine.cpp +++ b/be/src/storage/storage_engine.cpp @@ -140,17 +140,15 @@ int64_t BaseStorageEngine::memory_limitation_bytes_per_thread_for_schema_change( } void BaseStorageEngine::_start_adaptive_thread_controller() { + if (!config::enable_adaptive_flush_threads) { + return; + } + auto* system_metrics = DorisMetrics::instance()->system_metrics(); auto* s3_upload_pool = ExecEnv::GetInstance()->s3_file_upload_thread_pool(); _adaptive_thread_controller.init(system_metrics, s3_upload_pool); - // Cloud write backpressure also uses is_io_busy(), so initialize the IO signal even when - // adaptive flush-thread adjustment is disabled. - if (!config::enable_adaptive_flush_threads) { - return; - } - if (_memtable_flush_executor) { auto* flush_pool = _memtable_flush_executor->flush_pool(); auto* high_prio_pool = _memtable_flush_executor->high_prio_flush_pool(); diff --git a/be/test/storage/adaptive_thread_pool_controller_test.cpp b/be/test/storage/adaptive_thread_pool_controller_test.cpp index 38767812d48cda..6c4e42fea05897 100644 --- a/be/test/storage/adaptive_thread_pool_controller_test.cpp +++ b/be/test/storage/adaptive_thread_pool_controller_test.cpp @@ -22,12 +22,10 @@ #include #include -#include "cloud/config.h" #include "common/config.h" #include "common/metrics/metrics.h" #include "common/metrics/system_metrics.h" #include "testutil/test_util.h" -#include "util/countdown_latch.h" #include "util/threadpool.h" namespace doris { @@ -46,8 +44,6 @@ class AdaptiveThreadPoolControllerTest : public testing::Test { protected: void SetUp() override { _original_enable_adaptive = config::enable_adaptive_flush_threads; - _original_deploy_mode = config::deploy_mode; - _original_cloud_unique_id = config::cloud_unique_id; int num_cpus = std::thread::hardware_concurrency(); if (num_cpus <= 0) num_cpus = 1; @@ -68,15 +64,11 @@ class AdaptiveThreadPoolControllerTest : public testing::Test { void TearDown() override { config::enable_adaptive_flush_threads = _original_enable_adaptive; - config::deploy_mode = _original_deploy_mode; - config::cloud_unique_id = _original_cloud_unique_id; if (_pool) _pool->shutdown(); if (_pool2) _pool2->shutdown(); } bool _original_enable_adaptive; - std::string _original_deploy_mode; - std::string _original_cloud_unique_id; std::unique_ptr _pool; std::unique_ptr _pool2; }; @@ -268,44 +260,6 @@ TEST_F(AdaptiveThreadPoolControllerTest, TestIoBusyCpuBusyWithNullMetrics) { EXPECT_FALSE(controller.is_cpu_busy()); } -TEST_F(AdaptiveThreadPoolControllerTest, TestCloudIoBusyUsesS3UploadQueue) { - config::deploy_mode = "cloud"; - config::cloud_unique_id.clear(); - - std::unique_ptr upload_pool; - ASSERT_TRUE(ThreadPoolBuilder("TestS3UploadPool") - .set_min_threads(1) - .set_max_threads(1) - .set_max_queue_size(AdaptiveThreadPoolController::kS3QueueBusyThreshold + 1) - .build(&upload_pool) - .ok()); - - CountDownLatch task_started(1); - CountDownLatch release_task(1); - ASSERT_TRUE(upload_pool - ->submit_func([&]() { - task_started.count_down(); - release_task.wait(); - }) - .ok()); - EXPECT_TRUE(task_started.wait_for(std::chrono::seconds(5))); - - for (int i = 0; i < AdaptiveThreadPoolController::kS3QueueBusyThreshold; ++i) { - EXPECT_TRUE(upload_pool->submit_func([]() {}).ok()); - } - - AdaptiveThreadPoolController controller; - controller.init(nullptr, upload_pool.get()); - EXPECT_FALSE(controller.is_io_busy()); - - EXPECT_TRUE(upload_pool->submit_func([]() {}).ok()); - EXPECT_TRUE(controller.is_io_busy()); - - release_task.count_down(); - upload_pool->shutdown(); - upload_pool->wait(); -} - TEST_F(AdaptiveThreadPoolControllerTest, TestCpuBusyUsesCpuMetricsDelta) { MetricRegistry registry("test"); const std::string before_path = get_stat_test_data_path("stat_cpu_busy_before"); From 32e39fc268f490d8685590d70ab86c30a31740af Mon Sep 17 00:00:00 2001 From: laihui Date: Mon, 17 Aug 2026 18:51:51 +0800 Subject: [PATCH 3/3] [fix](be) Separate local and cloud flush backpressure Use the doubled physical flush task limit only for local row-binlog group writers. In cloud mode, replace flush-count backpressure with the S3 upload queue busy signal and initialize that signal independently of adaptive flush-thread adjustment. --- be/src/cloud/cloud_delta_writer.cpp | 6 +-- be/src/load/delta_writer/delta_writer.cpp | 3 +- .../load/delta_writer/delta_writer_context.h | 6 --- be/src/storage/storage_engine.cpp | 10 ++-- .../load/delta_writer/delta_writer_test.cpp | 11 ----- .../adaptive_thread_pool_controller_test.cpp | 46 +++++++++++++++++++ 6 files changed, 57 insertions(+), 25 deletions(-) diff --git a/be/src/cloud/cloud_delta_writer.cpp b/be/src/cloud/cloud_delta_writer.cpp index 7b6e2785fca3d0..bb6fdcf26b9f67 100644 --- a/be/src/cloud/cloud_delta_writer.cpp +++ b/be/src/cloud/cloud_delta_writer.cpp @@ -99,9 +99,9 @@ Status CloudDeltaWriter::write(const Block* block, const TabletAddRowsPayload& r CHECK(_is_init || _is_cancelled); { SCOPED_TIMER(_wait_flush_limit_timer); - const auto effective_flush_running_count_limit = - config::memtable_flush_running_count_limit * _req.memtable_flush_task_count(); - while (_memtable_writer->flush_running_count() >= effective_flush_running_count_limit) { + auto* adaptive_controller = + ExecEnv::GetInstance()->storage_engine().adaptive_thread_controller(); + while (adaptive_controller->is_io_busy()) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } diff --git a/be/src/load/delta_writer/delta_writer.cpp b/be/src/load/delta_writer/delta_writer.cpp index 9dea3d8154fc66..abcda75afa4762 100644 --- a/be/src/load/delta_writer/delta_writer.cpp +++ b/be/src/load/delta_writer/delta_writer.cpp @@ -179,7 +179,8 @@ Status DeltaWriter::write(const Block* block, const TabletAddRowsPayload& rows, { SCOPED_TIMER(_wait_flush_limit_timer); const auto effective_flush_running_count_limit = - config::memtable_flush_running_count_limit * _req.memtable_flush_task_count(); + 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)); } diff --git a/be/src/load/delta_writer/delta_writer_context.h b/be/src/load/delta_writer/delta_writer_context.h index 0a078c9a7420b3..856601c0be6a83 100644 --- a/be/src/load/delta_writer/delta_writer_context.h +++ b/be/src/load/delta_writer/delta_writer_context.h @@ -56,12 +56,6 @@ struct WriteRequest { WriteRequestType write_req_type = WriteRequestType::DATA; std::string storage_vault_id; bool enable_table_memtable_backpressure = false; - - int32_t memtable_flush_task_count() const { - // A GROUP request flushes both the data and row-binlog memtables. The two physical - // flush tasks are counted independently by flush_running_count(). - return write_req_type == WriteRequestType::GROUP ? 2 : 1; - } }; struct TabletAddRowsPayload { diff --git a/be/src/storage/storage_engine.cpp b/be/src/storage/storage_engine.cpp index 2093a17e663ae8..80f92e4b14435f 100644 --- a/be/src/storage/storage_engine.cpp +++ b/be/src/storage/storage_engine.cpp @@ -140,15 +140,17 @@ int64_t BaseStorageEngine::memory_limitation_bytes_per_thread_for_schema_change( } void BaseStorageEngine::_start_adaptive_thread_controller() { - if (!config::enable_adaptive_flush_threads) { - return; - } - auto* system_metrics = DorisMetrics::instance()->system_metrics(); auto* s3_upload_pool = ExecEnv::GetInstance()->s3_file_upload_thread_pool(); _adaptive_thread_controller.init(system_metrics, s3_upload_pool); + // Cloud write backpressure also uses is_io_busy(), so initialize the IO signal even when + // adaptive flush-thread adjustment is disabled. + if (!config::enable_adaptive_flush_threads) { + return; + } + if (_memtable_flush_executor) { auto* flush_pool = _memtable_flush_executor->flush_pool(); auto* high_prio_pool = _memtable_flush_executor->high_prio_flush_pool(); diff --git a/be/test/load/delta_writer/delta_writer_test.cpp b/be/test/load/delta_writer/delta_writer_test.cpp index a7b75f0bc94019..62eafce41135f6 100644 --- a/be/test/load/delta_writer/delta_writer_test.cpp +++ b/be/test/load/delta_writer/delta_writer_test.cpp @@ -66,17 +66,6 @@ namespace doris { class OlapMeta; -TEST(WriteRequestTest, MemtableFlushTaskCount) { - WriteRequest req; - EXPECT_EQ(req.memtable_flush_task_count(), 1); - - req.write_req_type = WriteRequestType::ROW_BINLOG; - EXPECT_EQ(req.memtable_flush_task_count(), 1); - - req.write_req_type = WriteRequestType::GROUP; - EXPECT_EQ(req.memtable_flush_task_count(), 2); -} - // This is DeltaWriter unit test which used by streaming load. // And also it should take schema change into account after streaming load. diff --git a/be/test/storage/adaptive_thread_pool_controller_test.cpp b/be/test/storage/adaptive_thread_pool_controller_test.cpp index 6c4e42fea05897..38767812d48cda 100644 --- a/be/test/storage/adaptive_thread_pool_controller_test.cpp +++ b/be/test/storage/adaptive_thread_pool_controller_test.cpp @@ -22,10 +22,12 @@ #include #include +#include "cloud/config.h" #include "common/config.h" #include "common/metrics/metrics.h" #include "common/metrics/system_metrics.h" #include "testutil/test_util.h" +#include "util/countdown_latch.h" #include "util/threadpool.h" namespace doris { @@ -44,6 +46,8 @@ class AdaptiveThreadPoolControllerTest : public testing::Test { protected: void SetUp() override { _original_enable_adaptive = config::enable_adaptive_flush_threads; + _original_deploy_mode = config::deploy_mode; + _original_cloud_unique_id = config::cloud_unique_id; int num_cpus = std::thread::hardware_concurrency(); if (num_cpus <= 0) num_cpus = 1; @@ -64,11 +68,15 @@ class AdaptiveThreadPoolControllerTest : public testing::Test { void TearDown() override { config::enable_adaptive_flush_threads = _original_enable_adaptive; + config::deploy_mode = _original_deploy_mode; + config::cloud_unique_id = _original_cloud_unique_id; if (_pool) _pool->shutdown(); if (_pool2) _pool2->shutdown(); } bool _original_enable_adaptive; + std::string _original_deploy_mode; + std::string _original_cloud_unique_id; std::unique_ptr _pool; std::unique_ptr _pool2; }; @@ -260,6 +268,44 @@ TEST_F(AdaptiveThreadPoolControllerTest, TestIoBusyCpuBusyWithNullMetrics) { EXPECT_FALSE(controller.is_cpu_busy()); } +TEST_F(AdaptiveThreadPoolControllerTest, TestCloudIoBusyUsesS3UploadQueue) { + config::deploy_mode = "cloud"; + config::cloud_unique_id.clear(); + + std::unique_ptr upload_pool; + ASSERT_TRUE(ThreadPoolBuilder("TestS3UploadPool") + .set_min_threads(1) + .set_max_threads(1) + .set_max_queue_size(AdaptiveThreadPoolController::kS3QueueBusyThreshold + 1) + .build(&upload_pool) + .ok()); + + CountDownLatch task_started(1); + CountDownLatch release_task(1); + ASSERT_TRUE(upload_pool + ->submit_func([&]() { + task_started.count_down(); + release_task.wait(); + }) + .ok()); + EXPECT_TRUE(task_started.wait_for(std::chrono::seconds(5))); + + for (int i = 0; i < AdaptiveThreadPoolController::kS3QueueBusyThreshold; ++i) { + EXPECT_TRUE(upload_pool->submit_func([]() {}).ok()); + } + + AdaptiveThreadPoolController controller; + controller.init(nullptr, upload_pool.get()); + EXPECT_FALSE(controller.is_io_busy()); + + EXPECT_TRUE(upload_pool->submit_func([]() {}).ok()); + EXPECT_TRUE(controller.is_io_busy()); + + release_task.count_down(); + upload_pool->shutdown(); + upload_pool->wait(); +} + TEST_F(AdaptiveThreadPoolControllerTest, TestCpuBusyUsesCpuMetricsDelta) { MetricRegistry registry("test"); const std::string before_path = get_stat_test_data_path("stat_cpu_busy_before");