From 2a8bff6ce59c8f4a97e9b1d02cebe97f95e996d6 Mon Sep 17 00:00:00 2001 From: meiyi Date: Thu, 23 Jul 2026 11:02:14 +0800 Subject: [PATCH] [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 + .../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 ++++++ ...t_group_commit_wal_num_backpressure.groovy | 99 +++++++++++++++++++ 11 files changed, 216 insertions(+), 22 deletions(-) create mode 100644 regression-test/suites/insert_p0/group_commit/test_group_commit_wal_num_backpressure.groovy diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 59dffdf1ac34e3..8f72cdbacd9f42 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1419,6 +1419,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 033b755409cc64..8dc1e08bc093ed 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1506,6 +1506,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 3f5dac3cd7bbbe..20138cb8062d20 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()) { + 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 3a152a25b16730..57b42ce4c08cb5 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& load_block_queue, int be_exe_version, - std::shared_ptr create_plan_dep, std::shared_ptr put_block_dep) { + TGroupCommitMode::type group_commit_mode, std::shared_ptr create_plan_dep, + std::shared_ptr 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("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("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(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( + "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& load_block_queue, - int be_exe_version, - std::shared_ptr create_plan_dep, - std::shared_ptr 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& load_block_queue, + int be_exe_version, TGroupCommitMode::type group_commit_mode, + std::shared_ptr create_plan_dep, std::shared_ptr put_block_dep) { std::shared_ptr 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 5ab3831e4f5cc4..895715afdeedc1 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 @@ class GroupCommitTable { 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& load_block_queue, - int be_exe_version, + int be_exe_version, TGroupCommitMode::type group_commit_mode, std::shared_ptr create_plan_dep, std::shared_ptr put_block_dep); Status get_load_block_queue(const TUniqueId& instance_id, @@ -177,6 +177,7 @@ class GroupCommitTable { 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& mem_tracker, std::shared_ptr& created_load_block_queue); @@ -222,7 +223,7 @@ class GroupCommitMgr { 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& load_block_queue, - int be_exe_version, + int be_exe_version, TGroupCommitMode::type group_commit_mode, std::shared_ptr create_plan_dep, std::shared_ptr 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 06d009404f7efa..e7962cc3396ce1 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 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 4157cc11d19f4e..48985e0cba8074 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 @@ class WalManager { 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 8edb1d938c5238..13b6dc1c67f485 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 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 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& 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 89223c65668e0d..57a0acfb6cf4ec 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 @@ class WalTable { 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 @@ class WalTable { // key is wal_path std::map> _replay_wal_map; std::list> _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 717851d1f098b1..a6dabd57bdeb6b 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 00000000000000..478fc2c71566d1 --- /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) +}