This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 9e8c558c7dd branch-4.1: [fix](be) Backpressure async group commit by
table WAL count #65362 (#65940)
9e8c558c7dd is described below
commit 9e8c558c7ddecc41c68c38da3e701011a6f35d44
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Jul 23 20:22:08 2026 +0800
branch-4.1: [fix](be) Backpressure async group commit by table WAL count
#65362 (#65940)
Cherry-picked from #65362
Co-authored-by: meiyi <[email protected]>
---
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 1beaf56941f..3ec3fc0b249 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1422,6 +1422,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 e7b43146d68..7daad1f3a1b 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1509,6 +1509,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 3f5dac3cd7b..20138cb8062 100644
--- a/be/src/exec/operator/group_commit_block_sink_operator.cpp
+++ b/be/src/exec/operator/group_commit_block_sink_operator.cpp
@@ -77,7 +77,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();
}
@@ -87,8 +90,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 3a152a25b16..57b42ce4c08 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"
@@ -268,16 +269,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 &&
@@ -293,11 +297,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();
@@ -308,7 +314,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() {
@@ -841,13 +871,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);
@@ -860,7 +888,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 8edb1d938c5..13b6dc1c67f 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 717851d1f09..a6dabd57bde 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 {
@@ -381,6 +383,36 @@ TEST_F(WalManagerTest, 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]
