[fix](be) Backpressure async group commit by table WAL count

be/src/common/config.cpp

@@ -1435,6 +1435,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");

be/src/common/config.h

@@ -1519,6 +1519,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);
 

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();

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();
 }
 

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<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 @@ 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<MemTrackerLimiter>& mem_tracker,
  std::shared_ptr<LoadBlockQueue>& 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<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);

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) {

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

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;

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<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
+} // namespace doris

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;