This is an automated email from the ASF dual-hosted git repository.
mymeiyi 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 0c2a7b5a7cf [fix](be) Backpressure async group commit by table WAL
count (#65362)
0c2a7b5a7cf is described below
commit 0c2a7b5a7cf3a2c4bf74d3e99341de98fa5fd01f
Author: meiyi <[email protected]>
AuthorDate: Thu Jul 23 11:02:14 2026 +0800
[fix](be) Backpressure async group commit by table WAL count (#65362)
Problem Summary: Async group commit WAL replay failures can leave table
WAL backlog growing while later stream loads continue to be admitted.
This adds a per-table group commit WAL count limit, tracks WAL queue
size in WalManager, records replay failure reasons for diagnostics, and
rejects new async group commit loads once the backlog reaches the
configured limit.
---
be/src/common/config.cpp | 3 +
be/src/common/config.h | 2 +
.../operator/group_commit_block_sink_operator.cpp | 9 +-
be/src/load/group_commit/group_commit_mgr.cpp | 56 +++++++++---
be/src/load/group_commit/group_commit_mgr.h | 5 +-
be/src/load/group_commit/wal/wal_manager.cpp | 13 ++-
be/src/load/group_commit/wal/wal_manager.h | 1 +
be/src/load/group_commit/wal/wal_table.cpp | 14 +++
be/src/load/group_commit/wal/wal_table.h | 4 +-
be/test/format/wal/wal_manager_test.cpp | 32 +++++++
.../test_group_commit_wal_num_backpressure.groovy | 99 ++++++++++++++++++++++
11 files changed, 216 insertions(+), 22 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 4d86958fb73..3955878e276 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1450,6 +1450,9 @@ DEFINE_mInt32(group_commit_queue_mem_limit, "67108864");
// group_commit_wal_max_disk_limit=1024 or group_commit_wal_max_disk_limit=10%
can be automatically identified.
DEFINE_String(group_commit_wal_max_disk_limit, "10%");
DEFINE_Bool(group_commit_wait_replay_wal_finish, "false");
+// Max WAL count for one table before rejecting async group commit loads.
+// 0 means no limit.
+DEFINE_mInt32(group_commit_max_wal_num_per_table, "10");
// Max time(ms) to wait for creating group commit plan fragment.
// 0 means no timeout, default 2min.
DEFINE_mInt32(group_commit_create_plan_timeout_ms, "120000");
diff --git a/be/src/common/config.h b/be/src/common/config.h
index 85f6a2bf146..cb09dfcb538 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1528,6 +1528,8 @@ DECLARE_mInt32(group_commit_queue_mem_limit);
// group_commit_wal_max_disk_limit=1024 or group_commit_wal_max_disk_limit=10%
can be automatically identified.
DECLARE_mString(group_commit_wal_max_disk_limit);
DECLARE_Bool(group_commit_wait_replay_wal_finish);
+// Max WAL count for one table before rejecting async group commit loads. 0
means no limit.
+DECLARE_mInt32(group_commit_max_wal_num_per_table);
// Max time(ms) to wait for creating group commit plan fragment. 0 means no
timeout.
DECLARE_mInt32(group_commit_create_plan_timeout_ms);
diff --git a/be/src/exec/operator/group_commit_block_sink_operator.cpp
b/be/src/exec/operator/group_commit_block_sink_operator.cpp
index 66cf4aaee06..2119cf40935 100644
--- a/be/src/exec/operator/group_commit_block_sink_operator.cpp
+++ b/be/src/exec/operator/group_commit_block_sink_operator.cpp
@@ -76,7 +76,10 @@ Status GroupCommitBlockSinkLocalState::open(RuntimeState*
state) {
"CreateGroupCommitPlanDependency", true);
_put_block_dependency = Dependency::create_shared(_parent->operator_id(),
_parent->node_id(),
"GroupCommitPutBlockDependency", true);
- [[maybe_unused]] auto st = _initialize_load_queue();
+ auto st = _initialize_load_queue();
+ if (st.is<ErrorCode::EXCEEDED_LIMIT>()) {
+ return st;
+ }
return Status::OK();
}
@@ -86,8 +89,8 @@ Status
GroupCommitBlockSinkLocalState::_initialize_load_queue() {
if (_state->exec_env()->wal_mgr()->is_running()) {
RETURN_IF_ERROR(_state->exec_env()->group_commit_mgr()->get_first_block_load_queue(
p._db_id, p._table_id, p._base_schema_version,
p._schema->indexes().size(),
- p._load_id, _load_block_queue, _state->be_exec_version(),
_create_plan_dependency,
- _put_block_dependency));
+ p._load_id, _load_block_queue, _state->be_exec_version(),
_group_commit_mode,
+ _create_plan_dependency, _put_block_dependency));
_state->set_import_label(_load_block_queue->label);
_state->set_wal_id(_load_block_queue->txn_id); // wal_id is txn_id
return Status::OK();
diff --git a/be/src/load/group_commit/group_commit_mgr.cpp
b/be/src/load/group_commit/group_commit_mgr.cpp
index 930c518ae9a..b0dd97ed130 100644
--- a/be/src/load/group_commit/group_commit_mgr.cpp
+++ b/be/src/load/group_commit/group_commit_mgr.cpp
@@ -31,6 +31,7 @@
#include "runtime/fragment_mgr.h"
#include "runtime/memory/mem_tracker_limiter.h"
#include "runtime/thread_context.h"
+#include "service/backend_options.h"
#include "util/client_cache.h"
#include "util/debug_points.h"
#include "util/thrift_rpc_helper.h"
@@ -267,16 +268,19 @@ void LoadBlockQueue::_cancel_without_lock(const Status&
st) {
Status GroupCommitTable::get_first_block_load_queue(
int64_t table_id, int64_t base_schema_version, int64_t index_size,
const UniqueId& load_id,
std::shared_ptr<LoadBlockQueue>& load_block_queue, int be_exe_version,
- std::shared_ptr<Dependency> create_plan_dep,
std::shared_ptr<Dependency> put_block_dep) {
+ TGroupCommitMode::type group_commit_mode, std::shared_ptr<Dependency>
create_plan_dep,
+ std::shared_ptr<Dependency> put_block_dep) {
DCHECK(table_id == _table_id);
std::unique_lock l(_lock);
- auto try_to_get_matched_queue = [&]() -> Status {
+ auto try_to_get_matched_queue = [&](bool& need_create_plan) -> Status {
+ need_create_plan = false;
for (const auto& [_, inner_block_queue] : _load_block_queues) {
if (inner_block_queue->contain_load_id(load_id)) {
load_block_queue = inner_block_queue;
return Status::OK();
}
}
+ RETURN_IF_ERROR(_check_wal_backlog(group_commit_mode));
for (const auto& [_, inner_block_queue] : _load_block_queues) {
if (!inner_block_queue->need_commit()) {
if (base_schema_version == inner_block_queue->schema_version &&
@@ -292,11 +296,13 @@ Status GroupCommitTable::get_first_block_load_queue(
}
}
}
- return Status::InternalError<false>("can not get a block queue for
table_id: " +
- std::to_string(_table_id) +
_create_plan_failed_reason);
+ need_create_plan = true;
+ return Status::OK();
};
- if (try_to_get_matched_queue().ok()) {
+ bool need_create_plan = false;
+ RETURN_IF_ERROR(try_to_get_matched_queue(need_create_plan));
+ if (!need_create_plan) {
return Status::OK();
}
create_plan_dep->block();
@@ -307,7 +313,31 @@ Status GroupCommitTable::get_first_block_load_queue(
_create_plan_deps.emplace(load_id, std::make_tuple(create_plan_dep,
put_block_dep,
base_schema_version,
index_size));
[[maybe_unused]] auto submit_st = _submit_create_group_commit_load();
- return try_to_get_matched_queue();
+ RETURN_IF_ERROR(try_to_get_matched_queue(need_create_plan));
+ if (need_create_plan) {
+ return Status::InternalError<false>("can not get a block queue for
table_id: " +
+ std::to_string(_table_id) +
_create_plan_failed_reason);
+ }
+ return Status::OK();
+}
+
+Status GroupCommitTable::_check_wal_backlog(TGroupCommitMode::type
group_commit_mode) {
+ int32_t max_wal_num = config::group_commit_max_wal_num_per_table;
+ if (group_commit_mode != TGroupCommitMode::ASYNC_MODE || max_wal_num <= 0)
{
+ return Status::OK();
+ }
+ size_t wal_num = _exec_env->wal_mgr()->get_wal_queue_size(_table_id);
+ if (wal_num < static_cast<size_t>(max_wal_num)) {
+ return Status::OK();
+ }
+ std::string failed_reason =
_exec_env->wal_mgr()->get_last_replay_wal_failed_reason(_table_id);
+ if (failed_reason.empty()) {
+ return Status::OK();
+ }
+ return Status::Error<ErrorCode::EXCEEDED_LIMIT>(
+ "Too many group commit async WALs for table {} on be host {}. wal
num={}, limit={}, "
+ "last replay wal failed reason: {}",
+ _table_id, BackendOptions::get_localhost(), wal_num, max_wal_num,
failed_reason);
}
Status GroupCommitTable::submit_create_group_commit_load() {
@@ -840,13 +870,11 @@ void GroupCommitMgr::_create_plan_worker() {
}
}
-Status GroupCommitMgr::get_first_block_load_queue(int64_t db_id, int64_t
table_id,
- int64_t base_schema_version,
int64_t index_size,
- const UniqueId& load_id,
-
std::shared_ptr<LoadBlockQueue>& load_block_queue,
- int be_exe_version,
- std::shared_ptr<Dependency>
create_plan_dep,
- std::shared_ptr<Dependency>
put_block_dep) {
+Status GroupCommitMgr::get_first_block_load_queue(
+ int64_t db_id, int64_t table_id, int64_t base_schema_version, int64_t
index_size,
+ const UniqueId& load_id, std::shared_ptr<LoadBlockQueue>&
load_block_queue,
+ int be_exe_version, TGroupCommitMode::type group_commit_mode,
+ std::shared_ptr<Dependency> create_plan_dep,
std::shared_ptr<Dependency> put_block_dep) {
std::shared_ptr<GroupCommitTable> group_commit_table;
{
std::lock_guard wlock(_lock);
@@ -859,7 +887,7 @@ Status GroupCommitMgr::get_first_block_load_queue(int64_t
db_id, int64_t table_i
}
RETURN_IF_ERROR(group_commit_table->get_first_block_load_queue(
table_id, base_schema_version, index_size, load_id,
load_block_queue, be_exe_version,
- create_plan_dep, put_block_dep));
+ group_commit_mode, create_plan_dep, put_block_dep));
return Status::OK();
}
diff --git a/be/src/load/group_commit/group_commit_mgr.h
b/be/src/load/group_commit/group_commit_mgr.h
index 5ab3831e4f5..895715afdee 100644
--- a/be/src/load/group_commit/group_commit_mgr.h
+++ b/be/src/load/group_commit/group_commit_mgr.h
@@ -166,7 +166,7 @@ public:
Status get_first_block_load_queue(int64_t table_id, int64_t
base_schema_version,
int64_t index_size, const UniqueId&
load_id,
std::shared_ptr<LoadBlockQueue>&
load_block_queue,
- int be_exe_version,
+ int be_exe_version,
TGroupCommitMode::type group_commit_mode,
std::shared_ptr<Dependency>
create_plan_dep,
std::shared_ptr<Dependency>
put_block_dep);
Status get_load_block_queue(const TUniqueId& instance_id,
@@ -177,6 +177,7 @@ public:
private:
Status _submit_create_group_commit_load();
+ Status _check_wal_backlog(TGroupCommitMode::type group_commit_mode);
Status _create_group_commit_load(int be_exe_version,
const std::shared_ptr<MemTrackerLimiter>&
mem_tracker,
std::shared_ptr<LoadBlockQueue>&
created_load_block_queue);
@@ -222,7 +223,7 @@ public:
Status get_first_block_load_queue(int64_t db_id, int64_t table_id, int64_t
base_schema_version,
int64_t index_size, const UniqueId&
load_id,
std::shared_ptr<LoadBlockQueue>&
load_block_queue,
- int be_exe_version,
+ int be_exe_version,
TGroupCommitMode::type group_commit_mode,
std::shared_ptr<Dependency>
create_plan_dep,
std::shared_ptr<Dependency>
put_block_dep);
void remove_load_id(int64_t table_id, const UniqueId& load_id);
diff --git a/be/src/load/group_commit/wal/wal_manager.cpp
b/be/src/load/group_commit/wal/wal_manager.cpp
index 06d009404f7..e7962cc3396 100644
--- a/be/src/load/group_commit/wal/wal_manager.cpp
+++ b/be/src/load/group_commit/wal/wal_manager.cpp
@@ -194,7 +194,7 @@ void WalManager::erase_wal_queue(int64_t table_id, int64_t
wal_id) {
}
size_t WalManager::get_wal_queue_size(int64_t table_id) {
- std::lock_guard<std::shared_mutex> wrlock(_wal_queue_lock);
+ std::shared_lock rdlock(_wal_queue_lock);
size_t count = 0;
if (table_id > 0) {
auto it = _wal_queues.find(table_id);
@@ -206,7 +206,7 @@ size_t WalManager::get_wal_queue_size(int64_t table_id) {
} else {
// table_id is -1 meaning get all table wal size
size_t max_count_per_table = 0;
- for (auto& [_, table_wals] : _wal_queues) {
+ for (const auto& [_, table_wals] : _wal_queues) {
size_t table_wal_count = table_wals.size();
count += table_wal_count;
if (table_wal_count > max_count_per_table) {
@@ -218,6 +218,15 @@ size_t WalManager::get_wal_queue_size(int64_t table_id) {
return count;
}
+std::string WalManager::get_last_replay_wal_failed_reason(int64_t table_id) {
+ std::shared_lock rdlock(_table_lock);
+ auto it = _table_map.find(table_id);
+ if (it != _table_map.end()) {
+ return it->second->get_last_replay_wal_failed_reason();
+ }
+ return "";
+}
+
Status WalManager::create_wal_path(int64_t db_id, int64_t table_id, int64_t
wal_id,
const std::string& label, std::string&
base_path,
uint32_t wal_version) {
diff --git a/be/src/load/group_commit/wal/wal_manager.h
b/be/src/load/group_commit/wal/wal_manager.h
index 4157cc11d19..48985e0cba8 100644
--- a/be/src/load/group_commit/wal/wal_manager.h
+++ b/be/src/load/group_commit/wal/wal_manager.h
@@ -80,6 +80,7 @@ public:
void add_wal_queue(int64_t table_id, int64_t wal_id);
void erase_wal_queue(int64_t table_id, int64_t wal_id);
size_t get_wal_queue_size(int64_t table_id);
+ std::string get_last_replay_wal_failed_reason(int64_t table_id);
// filename format:a_b_c_group_commit_xxx
// a:version
// b:be id
diff --git a/be/src/load/group_commit/wal/wal_table.cpp
b/be/src/load/group_commit/wal/wal_table.cpp
index 5dfd54354c3..060df525050 100644
--- a/be/src/load/group_commit/wal/wal_table.cpp
+++ b/be/src/load/group_commit/wal/wal_table.cpp
@@ -113,6 +113,12 @@ Status WalTable::_relay_wal_one_by_one() {
doris::wal_fail << 1;
LOG(WARNING) << "failed to replay wal=" << wal_info->get_wal_path()
<< ", st=" << st.to_string();
+ {
+ std::lock_guard<std::mutex> lock(_replay_wal_lock);
+ _last_replay_wal_failed_reason =
+ "failed to replay wal=" + wal_info->get_wal_path() +
+ ", st=" + st.to_string().substr(0, 100);
+ }
need_retry_wals.push_back(wal_info);
}
}
@@ -122,6 +128,9 @@ Status WalTable::_relay_wal_one_by_one() {
for (auto retry_wal_info : need_retry_wals) {
_replay_wal_map.emplace(retry_wal_info->get_wal_path(),
retry_wal_info);
}
+ if (_replay_wal_map.empty()) {
+ _last_replay_wal_failed_reason.clear();
+ }
}
return Status::OK();
}
@@ -308,6 +317,11 @@ size_t WalTable::size() {
return _replay_wal_map.size() + _replaying_queue.size();
}
+std::string WalTable::get_last_replay_wal_failed_reason() const {
+ std::lock_guard<std::mutex> lock(_replay_wal_lock);
+ return _last_replay_wal_failed_reason;
+}
+
Status WalTable::_get_column_info(int64_t db_id, int64_t tb_id,
std::map<int64_t, std::string>&
column_info_map) {
TGetColumnInfoRequest request;
diff --git a/be/src/load/group_commit/wal/wal_table.h
b/be/src/load/group_commit/wal/wal_table.h
index 89223c65668..57a0acfb6cf 100644
--- a/be/src/load/group_commit/wal/wal_table.h
+++ b/be/src/load/group_commit/wal/wal_table.h
@@ -40,6 +40,7 @@ public:
Status replay_wals();
size_t size();
void stop();
+ std::string get_last_replay_wal_failed_reason() const;
private:
void _pick_relay_wals();
@@ -67,5 +68,6 @@ private:
// key is wal_path
std::map<std::string, std::shared_ptr<WalInfo>> _replay_wal_map;
std::list<std::shared_ptr<WalInfo>> _replaying_queue;
+ std::string _last_replay_wal_failed_reason;
};
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/test/format/wal/wal_manager_test.cpp
b/be/test/format/wal/wal_manager_test.cpp
index 2189a0d33df..7bdafc58cfa 100644
--- a/be/test/format/wal/wal_manager_test.cpp
+++ b/be/test/format/wal/wal_manager_test.cpp
@@ -33,6 +33,8 @@
#include "runtime/memory/mem_tracker.h"
#include "runtime/runtime_state.h"
#include "runtime/user_function_cache.h"
+#include "util/debug_points.h"
+#include "util/defer_op.h"
namespace doris {
@@ -382,6 +384,36 @@ TEST_F(WalManagerTest,
DISABLED_read_block_fail_with_not_equal) {
WARN_IF_ERROR(scanner->close(&_runtime_state), "fail to close scanner");
}
+TEST_F(WalManagerTest, TestLastReplayWalFailedReason) {
+ const auto origin_enable_debug_points = config::enable_debug_points;
+ config::enable_debug_points = true;
+ DebugPoints::instance()->add("WalTable.replay_wals.stop");
+ Defer defer([origin_enable_debug_points]() {
+ DebugPoints::instance()->remove("WalTable.replay_wals.stop");
+ config::enable_debug_points = origin_enable_debug_points;
+ });
+
+ const int64_t wal_id = 789;
+ const std::string label = "test_last_replay_failed_reason";
+ const std::string wal_path = _wal_dir + "/" + std::to_string(_db_id) + "/"
+
+ std::to_string(_tb_id) + "/" +
std::to_string(_version_1) + "_" +
+ std::to_string(_backend_id) + "_" +
std::to_string(wal_id) + "_" +
+ label;
+
std::filesystem::copy_file("./be/test/exec/test_data/wal_scanner/wal_version1",
wal_path,
+
std::filesystem::copy_options::overwrite_existing);
+
+ WalTable wal_table(_env, _db_id, _tb_id);
+ wal_table.add_wal(wal_id, wal_path);
+ EXPECT_EQ(wal_table.replay_wals(), Status::OK());
+ auto failed_reason = wal_table.get_last_replay_wal_failed_reason();
+ EXPECT_NE(failed_reason.find("WalTable.replay_wals.stop"),
failed_reason.npos);
+ EXPECT_NE(failed_reason.find(wal_path), failed_reason.npos);
+
+ DebugPoints::instance()->remove("WalTable.replay_wals.stop");
+ EXPECT_EQ(wal_table.replay_wals(), Status::OK());
+ EXPECT_TRUE(wal_table.get_last_replay_wal_failed_reason().empty());
+}
+
TEST_F(WalManagerTest, TestDynamicWalSpaceLimt) {
// 1T
size_t available_bytes = 1099511627776;
diff --git
a/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_num_backpressure.groovy
b/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_num_backpressure.groovy
new file mode 100644
index 00000000000..478fc2c7156
--- /dev/null
+++
b/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_num_backpressure.groovy
@@ -0,0 +1,99 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_group_commit_wal_num_backpressure", "nonConcurrent") {
+ def getRowCount = { expectedRowCount ->
+ Awaitility.await().atMost(60, SECONDS).pollInterval(1, SECONDS).until(
+ {
+ def result = sql "select count(*) from
test_group_commit_wal_num_backpressure"
+ logger.info("table: test_group_commit_wal_num_backpressure,
rowCount: ${result}, expectedRowCount: ${expectedRowCount}")
+ return result[0][0] == expectedRowCount
+ }
+ )
+ }
+
+ sql """ DROP TABLE IF EXISTS test_group_commit_wal_num_backpressure """
+ sql """
+ CREATE TABLE IF NOT EXISTS test_group_commit_wal_num_backpressure (
+ `k` int,
+ `v` int
+ ) engine=olap
+ DISTRIBUTED BY HASH(`k`)
+ BUCKETS 1
+ properties(
+ "replication_num" = "1",
+ "group_commit_interval_ms" = "10000",
+ "group_commit_data_bytes" = "1"
+ )
+ """
+
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ GetDebugPoint().clearDebugPointsForAllFEs()
+ def rowCount = 0
+ try {
+ setBeConfigTemporary([group_commit_max_wal_num_per_table: 5]) {
+
GetDebugPoint().enableDebugPointForAllBEs("LoadBlockQueue._finish_group_commit_load.load_error")
+
GetDebugPoint().enableDebugPointForAllBEs("WalTable::_handle_stream_load.fail")
+
+ def backendIps = [:]
+ def backendHttpPorts = [:]
+ getBackendIpHttpPort(backendIps, backendHttpPorts)
+ def backendId = backendIps.keySet()[0]
+ def beHost = backendIps.get(backendId)
+ def beHttpPort = backendHttpPorts.get(backendId) as int
+
+ def streamLoadToBe = {
+ streamLoad {
+ table "test_group_commit_wal_num_backpressure"
+ set 'column_separator', ','
+ set 'group_commit', 'async_mode'
+ unset 'label'
+ file 'group_commit_wal_msg.csv'
+ time 10000
+ directToBe beHost, beHttpPort
+ }
+ rowCount += 5
+ }
+
+ def blocked = false
+ def maxAttempts = 100
+ for (int i = 0; i < maxAttempts && !blocked; ++i) {
+ try {
+ streamLoadToBe()
+ sleep(i < 10 ? 100 : 1000)
+ } catch (Exception e) {
+ logger.info("catch expected exception: " + e.getMessage())
+ assertTrue(e.getMessage().contains("Too many group commit
async WALs"))
+ assertTrue(e.getMessage().contains("limit=5"))
+ assertTrue(e.getMessage().contains("last replay wal failed
reason"))
+
assertTrue(e.getMessage().contains("WalTable::_handle_stream_load.fail"))
+ blocked = true
+ break
+ }
+ }
+ assertTrue(blocked)
+ }
+ } finally {
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ GetDebugPoint().clearDebugPointsForAllFEs()
+ }
+
+ // getRowCount(rowCount)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
