[improvement](be) Add table-level memtable flush backpressure

### What problem does this PR solve?

Issue Number: None

Related PR: #62661

Problem Summary: Adaptive random bucket load can rotate writes across multiple tablets on the same BE. When one table generates too many memtables faster than flush can drain them, new write batches should be throttled at table granularity instead of only relying on per-writer or process memory limits. This change tracks pending WRITE_FINISHED/FLUSH memtables by table id in MemTableMemoryLimiter and blocks new DeltaWriter writes when the per-BE table pending count reaches the configured limit.

### Release note

Add BE configs enable_table_memtable_flush_backpressure and table_memtable_flush_pending_count_limit to throttle load writes when one table has too many pending flush memtables on a BE.

### Check List (For Author)

- Test: Manual test
    - git diff --check
    - build-support/clang-format.sh attempted but failed because llvm@16 is not installed in this environment
    - ./build.sh --be attempted but failed during CMake configuration before Doris source compilation because OpenMP_C was not found for OpenBLAS
- Behavior changed: Yes. When enable_table_memtable_flush_backpressure is true, writes to a table block once its pending flush memtable count on a BE reaches table_memtable_flush_pending_count_limit.
- Does this need documentation: No

Author: sollhui

be/src/cloud/cloud_delta_writer.cpp

@@ -22,6 +22,8 @@
 #include "cloud/cloud_storage_engine.h"
 #include "cloud/config.h"
 #include "load/delta_writer/delta_writer.h"
+#include "load/memtable/memtable_memory_limiter.h"
+#include "runtime/exec_env.h"
 #include "runtime/thread_context.h"
 
 namespace doris {
@@ -72,6 +74,8 @@ Status CloudDeltaWriter::write(const Block* block, const DorisVector<uint32_t>&
     if (row_idxs.empty()) [[unlikely]] {
         return Status::OK();
     }
+    ExecEnv::GetInstance()->memtable_memory_limiter()->handle_table_memtable_backpressure(
+            nullptr, table_id());
     std::lock_guard lock(_mtx);
     CHECK(_is_init || _is_cancelled);
     {

be/src/common/config.cpp

@@ -842,6 +842,10 @@ DEFINE_mBool(enable_adaptive_flush_threads, "true");
 
 // Whether to enable queue-based backpressure in MemTableMemoryLimiter.
 DEFINE_mBool(enable_memtable_flush_queue_backpressure, "false");
+// Whether to block writes when one table has too many pending flush memtables on this BE.
+DEFINE_mBool(enable_table_memtable_flush_backpressure, "true");
+// Max pending flush memtables for one table on this BE before blocking new writes.
+DEFINE_mInt32(table_memtable_flush_pending_count_limit, "10");
 
 // config for tablet meta checkpoint
 DEFINE_mInt32(tablet_meta_checkpoint_min_new_rowsets_num, "10");

be/src/common/config.h

@@ -900,6 +900,10 @@ DECLARE_mBool(enable_adaptive_flush_threads);
 // When enabled, writes will be throttled if the workload group's flush pool
 // queue size exceeds the threshold, even if memory limits are not reached.
 DECLARE_mBool(enable_memtable_flush_queue_backpressure);
+// Whether to block writes when one table has too many pending flush memtables on this BE.
+DECLARE_mBool(enable_table_memtable_flush_backpressure);
+// Max pending flush memtables for one table on this BE before blocking new writes.
+DECLARE_mInt32(table_memtable_flush_pending_count_limit);
 
 // config for tablet meta checkpoint
 DECLARE_mInt32(tablet_meta_checkpoint_min_new_rowsets_num);

be/src/load/delta_writer/delta_writer.cpp

@@ -36,6 +36,7 @@
 #include "core/block/block.h"
 #include "io/fs/file_writer.h" // IWYU pragma: keep
 #include "load/memtable/memtable_flush_executor.h"
+#include "load/memtable/memtable_memory_limiter.h"
 #include "runtime/exec_env.h"
 #include "runtime/thread_context.h"
 #include "service/backend_options.h"
@@ -124,6 +125,11 @@ void BaseDeltaWriter::set_tablet_load_rowset_num_info(
     collect_tablet_load_rowset_num_info(tablet, tablet_infos);
 }
 
+int64_t BaseDeltaWriter::table_id() const {
+    DORIS_CHECK(_req.table_schema_param != nullptr);
+    return _req.table_schema_param->table_id();
+}
+
 DeltaWriter::~DeltaWriter() = default;
 
 Status BaseDeltaWriter::init() {
@@ -152,6 +158,8 @@ Status DeltaWriter::write(const Block* block, const DorisVector<uint32_t>& row_i
     if (UNLIKELY(row_idxs.empty())) {
         return Status::OK();
     }
+    ExecEnv::GetInstance()->memtable_memory_limiter()->handle_table_memtable_backpressure(
+            nullptr, table_id());
     _lock_watch.start();
     std::lock_guard<std::mutex> l(_lock);
     _lock_watch.stop();

be/src/load/delta_writer/delta_writer.h

@@ -86,6 +86,8 @@ class BaseDeltaWriter {
 
     int64_t partition_id() const { return _req.partition_id; }
 
+    int64_t table_id() const;
+
     int64_t tablet_id() const { return _req.tablet_id; }
 
     int64_t txn_id() const { return _req.txn_id; }

be/src/load/delta_writer/delta_writer_v2.cpp

@@ -35,6 +35,7 @@
 #include "core/block/block.h"
 #include "exec/sink/load_stream_stub.h"
 #include "io/fs/file_writer.h" // IWYU pragma: keep
+#include "load/memtable/memtable_memory_limiter.h"
 #include "runtime/exec_env.h"
 #include "runtime/query_context.h"
 #include "service/backend_options.h"
@@ -93,6 +94,11 @@ DeltaWriterV2::~DeltaWriterV2() {
     static_cast<void>(_memtable_writer->cancel());
 }
 
+int64_t DeltaWriterV2::_table_id() const {
+    DORIS_CHECK(_req.table_schema_param != nullptr);
+    return _req.table_schema_param->table_id();
+}
+
 Status DeltaWriterV2::init() {
     if (_is_init) {
         return Status::OK();
@@ -148,6 +154,8 @@ Status DeltaWriterV2::write(const Block* block, const DorisVector<uint32_t>& row
     if (UNLIKELY(row_idxs.empty())) {
         return Status::OK();
     }
+    ExecEnv::GetInstance()->memtable_memory_limiter()->handle_table_memtable_backpressure(
+            [state = _state]() { return state->is_cancelled(); }, _table_id());
     _lock_watch.start();
     std::lock_guard<std::mutex> l(_lock);
     _lock_watch.stop();

be/src/load/delta_writer/delta_writer_v2.h

@@ -86,6 +86,7 @@ class DeltaWriterV2 {
     Status _build_current_tablet_schema(int64_t index_id,
                                         const OlapTableSchemaParam* table_schema_param,
                                         const TabletSchema& ori_tablet_schema);
+    int64_t _table_id() const;
 
     void _update_profile(RuntimeProfile* profile);
 

be/src/load/memtable/memtable_memory_limiter.cpp

@@ -36,6 +36,12 @@ DEFINE_GAUGE_METRIC_PROTOTYPE_5ARG(memtable_memory_limiter_mem_consumption, Metr
 
 bvar::LatencyRecorder g_memtable_memory_limit_latency_ms("mm_limiter_limit_time_ms");
 bvar::Adder<int> g_memtable_memory_limit_waiting_threads("mm_limiter_waiting_threads");
+bvar::LatencyRecorder g_memtable_table_backpressure_latency_ms(
+        "mm_limiter_table_backpressure_time_ms");
+bvar::Adder<int> g_memtable_table_backpressure_waiting_threads(
+        "mm_limiter_table_backpressure_waiting_threads");
+bvar::Status<int64_t> g_memtable_table_backpressure_pending_count(
+        "mm_limiter_table_backpressure_pending_count", 0);
 bvar::Status<int64_t> g_memtable_active_memory("mm_limiter_mem_active", 0);
 bvar::Status<int64_t> g_memtable_write_memory("mm_limiter_mem_write", 0);
 bvar::Status<int64_t> g_memtable_flush_memory("mm_limiter_mem_flush", 0);
@@ -123,6 +129,68 @@ int64_t MemTableMemoryLimiter::_need_flush() {
     return need_flush - _queue_mem_usage - _flush_mem_usage;
 }
 
+int64_t MemTableMemoryLimiter::_table_flush_pending_memtable_count(int64_t table_id) {
+    int64_t pending_memtables = 0;
+    for (const auto& writer : _writers) {
+        auto writer_sptr = writer.lock();
+        if (writer_sptr == nullptr) {
+            continue;
+        }
+        if (writer_sptr->table_id() == table_id) {
+            pending_memtables += writer_sptr->flush_pending_memtable_count();
+        }
+    }
+    return pending_memtables;
+}
+
+void MemTableMemoryLimiter::handle_table_memtable_backpressure(
+        std::function<bool()> cancel_check, int64_t table_id) {
+    if (!config::enable_table_memtable_flush_backpressure) {
+        return;
+    }
+    const int64_t pending_count_limit = config::table_memtable_flush_pending_count_limit;
+    if (pending_count_limit <= 0) {
+        return;
+    }
+    DORIS_CHECK(table_id > 0);
+
+    std::unique_lock<std::mutex> l(_lock);
+    int64_t pending_count = _table_flush_pending_memtable_count(table_id);
+    if (pending_count < pending_count_limit) {
+        return;
+    }
+
+    MonotonicStopWatch timer;
+    timer.start();
+    g_memtable_table_backpressure_waiting_threads << 1;
+    while (pending_count >= pending_count_limit) {
+        g_memtable_table_backpressure_pending_count.set_value(pending_count);
+        LOG_EVERY_T(INFO, 1) << "table memtable flush backpressure: table_id=" << table_id
+                              << ", pending_memtables=" << pending_count
+                              << ", limit=" << pending_count_limit
+                              << ", memtable writers num: " << _writers.size();
+        if (cancel_check && cancel_check()) {
+            LOG(INFO) << "cancelled when waiting for table memtable flush backpressure"
+                      << ", table_id=" << table_id << ", pending_memtables=" << pending_count
+                      << ", limit=" << pending_count_limit;
+            g_memtable_table_backpressure_waiting_threads << -1;
+            return;
+        }
+        static_cast<void>(_hard_limit_end_cond.wait_for(l, std::chrono::milliseconds(100)));
+        pending_count = _table_flush_pending_memtable_count(table_id);
+    }
+    g_memtable_table_backpressure_pending_count.set_value(pending_count);
+    g_memtable_table_backpressure_waiting_threads << -1;
+    timer.stop();
+    int64_t time_ms = timer.elapsed_time() / 1000 / 1000;
+    g_memtable_table_backpressure_latency_ms << time_ms;
+    LOG(INFO) << "waited " << PrettyPrinter::print(timer.elapsed_time(), TUnit::TIME_NS)
+              << " for table memtable flush backpressure"
+              << ", table_id=" << table_id << ", pending_memtables=" << pending_count
+              << ", limit=" << pending_count_limit
+              << ", memtable writers num: " << _writers.size();
+}
+
 void MemTableMemoryLimiter::handle_memtable_flush(std::function<bool()> cancel_check,
                                                    WorkloadGroup* wg) {
     ThreadPool* wg_flush_pool = (wg != nullptr) ? wg->get_memtable_flush_pool() : nullptr;

be/src/load/memtable/memtable_memory_limiter.h

@@ -45,6 +45,7 @@ class MemTableMemoryLimiter {
     // Every write operation will call this API to check if need flush memtable OR hang
     // when memory is not available.
     void handle_memtable_flush(std::function<bool()> cancel_check, WorkloadGroup* wg = nullptr);
+    void handle_table_memtable_backpressure(std::function<bool()> cancel_check, int64_t table_id);
 
     void register_writer(std::weak_ptr<MemTableWriter> writer);
 
@@ -64,6 +65,7 @@ class MemTableMemoryLimiter {
     bool _hard_limit_reached();
     bool _load_usage_low();
     int64_t _need_flush();
+    int64_t _table_flush_pending_memtable_count(int64_t table_id);
     int64_t _flush_active_memtables(int64_t need_flush);
     void _refresh_mem_tracker();
     std::mutex _lock;

be/src/load/memtable/memtable_writer.cpp

@@ -381,6 +381,27 @@ uint64_t MemTableWriter::flush_running_count() const {
     return _flush_token == nullptr ? 0 : _flush_token->get_stats().flush_running_count.load();
 }
 
+int64_t MemTableWriter::table_id() const {
+    DORIS_CHECK(_req.table_schema_param != nullptr);
+    return _req.table_schema_param->table_id();
+}
+
+int64_t MemTableWriter::flush_pending_memtable_count() {
+    std::lock_guard<std::mutex> l(_mem_table_ptr_lock);
+    int64_t memtable_count = 0;
+    for (const auto& mem_table : _freezed_mem_tables) {
+        auto mem_table_sptr = mem_table.lock();
+        if (mem_table_sptr == nullptr) {
+            continue;
+        }
+        auto mem_type = mem_table_sptr->get_mem_type();
+        if (mem_type == MemType::WRITE_FINISHED || mem_type == MemType::FLUSH) {
+            memtable_count++;
+        }
+    }
+    return memtable_count;
+}
+
 int64_t MemTableWriter::mem_consumption(MemType mem) {
     if (!_is_init) {
         // This method may be called before this writer is initialized.