[enhance](load) Schedule load tasks with per-load round robin and intra-load priority

be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp

@@ -26,8 +26,10 @@
 
 #include "cloud/cloud_meta_mgr.h"
 #include "cloud/cloud_tablet.h"
+#include "cloud/cloud_txn_delete_bitmap_cache.h"
 #include "common/status.h"
 #include "runtime/memory/mem_tracker_limiter.h"
+#include "storage/delete/calc_delete_bitmap_executor.h"
 #include "storage/olap_common.h"
 #include "storage/rowset/beta_rowset.h"
 #include "storage/rowset/rowset.h"
@@ -67,9 +69,7 @@ Status CloudEngineCalcDeleteBitmapTask::execute() {
     int64_t transaction_id = _cal_delete_bitmap_req.transaction_id;
     OlapStopWatch watch;
     VLOG_NOTICE << "begin to calculate delete bitmap. transaction_id=" << transaction_id;
-    std::unique_ptr<ThreadPoolToken> token =
-            _engine.calc_tablet_delete_bitmap_task_thread_pool().new_token(
-                    ThreadPool::ExecutionMode::CONCURRENT);
+    std::vector<std::unique_ptr<CalcDeleteBitmapToken>> tokens;
     DBUG_EXECUTE_IF("CloudEngineCalcDeleteBitmapTask.execute.enable_wait", {
         auto sleep_time = DebugPoints::instance()->get_debug_param_or_default<int32_t>(
                 "CloudEngineCalcDeleteBitmapTask.execute.enable_wait", "sleep_time", 3);
@@ -93,6 +93,19 @@ Status CloudEngineCalcDeleteBitmapTask::execute() {
             if (has_tablet_states) {
                 tablet_calc_delete_bitmap_ptr->set_tablet_state(partition.tablet_states[i]);
             }
+            // A partition's first subtransaction need not have written this tablet.
+            auto wg =
+                    _engine.txn_delete_bitmap_cache().get_workload_group(transaction_id, tablet_id);
+            for (auto sub_txn_id : partition.sub_txn_ids) {
+                if (wg) {
+                    break;
+                }
+                wg = _engine.txn_delete_bitmap_cache().get_workload_group(sub_txn_id, tablet_id);
+            }
+            auto& token =
+                    tokens.emplace_back(_engine.calc_delete_bitmap_executor()->create_load_token(
+                            transaction_id, LoadTaskPriority::HIGHEST, LoadTaskType::PARENT,
+                            std::move(wg)));
             const auto submit_time_us = MonotonicMicros();
             auto submit_st = token->submit_func(
                     [tablet_id, tablet_calc_delete_bitmap_ptr, this, submit_time_us]() {
@@ -104,16 +117,28 @@ Status CloudEngineCalcDeleteBitmapTask::execute() {
                             LOG(WARNING) << "handle calc delete bitmap fail, st=" << st.to_string();
                             add_error_tablet_id(tablet_id, st);
                         }
+                        return Status::OK();
                     });
             VLOG_DEBUG << "submit TabletCalcDeleteBitmapTask for tablet=" << tablet_id;
             if (!submit_st.ok()) {
-                _res = submit_st;
+                add_error_tablet_id(tablet_id, submit_st);
                 break;
             }
         }
     }
-    // wait for all finished
-    token->wait();
+    // Drain every submitted token before reading _res, which running tablet
+    // callbacks may still update. Keep the recorded tablet/submission error
+    // ahead of a generic cancellation reported by another token's wait().
+    Status wait_status;
+    for (auto& token : tokens) {
+        auto st = token->wait();
+        if (wait_status.ok() && !st.ok()) {
+            wait_status = st;
+        }
+    }
+    if (_res.ok()) {
+        _res = wait_status;
+    }
 
     LOG(INFO) << "finish to calculate delete bitmap on transaction."
               << "transaction_id=" << transaction_id << ", cost(us): " << watch.get_elapse_time_us()
@@ -149,7 +174,8 @@ void CloudTabletCalcDeleteBitmapTask::set_tablet_state(int64_t tablet_state) {
 Status CloudTabletCalcDeleteBitmapTask::handle(int64_t queue_time_us) const {
     VLOG_DEBUG << "start calculate delete bitmap on tablet " << _tablet_id
                << ", txn_id=" << _transaction_id;
-    SCOPED_ATTACH_TASK(_mem_tracker);
+    // The bitmap token attaches the request context at the worker entry.
+    SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
     int64_t t1 = MonotonicMicros();
     auto base_tablet = DORIS_TRY(_engine.get_tablet(_tablet_id));
     auto get_tablet_time_us = MonotonicMicros() - t1;

be/src/cloud/cloud_meta_mgr.cpp

@@ -125,7 +125,9 @@ Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int
         });
 
         bthread_t bthread_id;
-        if (bthread_start_background(&bthread_id, nullptr, run_bthread_work, fn) != 0) {
+        if (SYNC_POINT_HOOK_RETURN_VALUE(
+                    bthread_start_background(&bthread_id, nullptr, run_bthread_work, fn),
+                    "bthread_fork_join::start_background") != 0) {
             run_bthread_work(fn);
         }
     }

be/src/cloud/cloud_rowset_builder.cpp

@@ -117,7 +117,8 @@ Status CloudRowsetBuilder::init() {
     _rowset_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false));
     _rowset_id = context.rowset_id;
 
-    _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor()->create_token();
+    _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor()->create_load_token(
+            _req.txn_id, LoadTaskPriority::HIGH, LoadTaskType::LEAF);
 
     if (!_skip_writing_rowset_metadata) {
         RETURN_IF_ERROR(_engine.meta_mgr().prepare_rowset(*_rowset_writer->rowset_meta(), "",
@@ -254,7 +255,7 @@ Status CloudRowsetBuilder::set_txn_related_info() {
         // For empty rowsets when skip_writing_empty_rowset_metadata=true,
         // store only a lightweight marker instead of full rowset info.
         // This allows CalcDeleteBitmapTask to detect and skip gracefully,
-        // while using minimal memory (~16 bytes per entry).
+        // while retaining the workload group for publish routing.
         if (_skip_writing_rowset_metadata) {
             _engine.txn_delete_bitmap_cache().mark_empty_rowset(_req.txn_id, _tablet->tablet_id(),
                                                                 _req.txn_expiration);

be/src/cloud/cloud_rowset_writer.cpp

@@ -89,7 +89,11 @@ Status CloudRowsetWriter::init(const RowsetWriterContext& rowset_writer_context)
     _context.segment_collector = std::make_shared<SegmentCollectorT<BaseBetaRowsetWriter>>(this);
     _context.file_writer_creator = std::make_shared<FileWriterCreatorT<BaseBetaRowsetWriter>>(this);
     if (_context.mow_context != nullptr) {
-        _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor_for_load()->create_token();
+        _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor()->create_load_token(
+                _context.txn_id,
+                _context.is_transient_rowset_writer ? LoadTaskPriority::HIGHEST
+                                                    : LoadTaskPriority::MID,
+                LoadTaskType::LEAF);
     }
     return Status::OK();
 }

be/src/cloud/cloud_storage_engine.cpp

@@ -220,14 +220,8 @@ Status CloudStorageEngine::open() {
 
     _calc_delete_bitmap_executor = std::make_unique<CalcDeleteBitmapExecutor>();
     _calc_delete_bitmap_executor->init("TabletCalcDeleteBitmapThreadPool",
-                                       config::calc_delete_bitmap_max_thread);
-
-    _calc_delete_bitmap_executor_for_load = std::make_unique<CalcDeleteBitmapExecutor>();
-    _calc_delete_bitmap_executor_for_load->init(
-            "LoadCalcDeleteBitmapThreadPool",
-            config::calc_delete_bitmap_for_load_max_thread > 0
-                    ? config::calc_delete_bitmap_for_load_max_thread
-                    : std::max(1, CpuInfo::num_cores() / 2));
+                                       config::calc_delete_bitmap_max_thread,
+                                       _memtable_flush_executor->flush_pool());
 
     // The default cache is set to 100MB, use memory limit to dynamic adjustment
     bool is_percent = false;
@@ -275,18 +269,15 @@ Status CloudStorageEngine::open() {
 
 #ifdef BE_TEST
 void CloudStorageEngine::init_calc_delete_bitmap_executor_for_UT() {
+    if (_memtable_flush_executor == nullptr) {
+        _memtable_flush_executor = std::make_unique<MemTableFlushExecutor>();
+        _memtable_flush_executor->init(1);
+    }
     if (_calc_delete_bitmap_executor == nullptr) {
         _calc_delete_bitmap_executor = std::make_unique<CalcDeleteBitmapExecutor>();
         _calc_delete_bitmap_executor->init("TabletCalcDeleteBitmapThreadPool",
-                                           config::calc_delete_bitmap_max_thread);
-    }
-    if (_calc_delete_bitmap_executor_for_load == nullptr) {
-        _calc_delete_bitmap_executor_for_load = std::make_unique<CalcDeleteBitmapExecutor>();
-        _calc_delete_bitmap_executor_for_load->init(
-                "LoadCalcDeleteBitmapThreadPool",
-                config::calc_delete_bitmap_for_load_max_thread > 0
-                        ? config::calc_delete_bitmap_for_load_max_thread
-                        : std::max(1, CpuInfo::num_cores() / 2));
+                                           config::calc_delete_bitmap_max_thread,
+                                           _memtable_flush_executor->flush_pool());
     }
 }
 #endif
@@ -317,9 +308,6 @@ void CloudStorageEngine::stop() {
     _adaptive_thread_controller.stop();
     LOG(INFO) << "Cloud storage engine is stopped.";
 
-    if (_calc_tablet_delete_bitmap_task_thread_pool) {
-        _calc_tablet_delete_bitmap_task_thread_pool->shutdown();
-    }
     if (_sync_delete_bitmap_thread_pool) {
         _sync_delete_bitmap_thread_pool->shutdown();
     }
@@ -391,11 +379,6 @@ Status CloudStorageEngine::start_bg_threads(std::shared_ptr<WorkloadGroup> wg_sp
             &_id_file_map_gc_thread));
     LOG(INFO) << "id file map gc thread started";
 
-    // add calculate tablet delete bitmap task thread pool
-    RETURN_IF_ERROR(ThreadPoolBuilder("TabletCalDeleteBitmapThreadPool")
-                            .set_min_threads(config::calc_tablet_delete_bitmap_task_max_thread)
-                            .set_max_threads(config::calc_tablet_delete_bitmap_task_max_thread)
-                            .build(&_calc_tablet_delete_bitmap_task_thread_pool));
     RETURN_IF_ERROR(ThreadPoolBuilder("SyncDeleteBitmapThreadPool")
                             .set_min_threads(config::sync_delete_bitmap_task_max_thread)
                             .set_max_threads(config::sync_delete_bitmap_task_max_thread)

be/src/cloud/cloud_storage_engine.h

@@ -97,9 +97,6 @@ class CloudStorageEngine final : public BaseStorageEngine {
 
     CloudCommittedRSMgr& committed_rs_mgr() const { return *_committed_rs_mgr; }
 
-    ThreadPool& calc_tablet_delete_bitmap_task_thread_pool() const {
-        return *_calc_tablet_delete_bitmap_task_thread_pool;
-    }
     ThreadPool& sync_delete_bitmap_thread_pool() const { return *_sync_delete_bitmap_thread_pool; }
 
     std::optional<StorageResource> get_storage_resource(const std::string& vault_id) {
@@ -234,7 +231,6 @@ class CloudStorageEngine final : public BaseStorageEngine {
     std::unique_ptr<CloudTabletMgr> _tablet_mgr;
     std::unique_ptr<CloudTxnDeleteBitmapCache> _txn_delete_bitmap_cache;
     std::unique_ptr<CloudCommittedRSMgr> _committed_rs_mgr;
-    std::unique_ptr<ThreadPool> _calc_tablet_delete_bitmap_task_thread_pool;
     std::unique_ptr<ThreadPool> _sync_delete_bitmap_thread_pool;
 
     // Components for cache warmup

be/src/cloud/cloud_tablets_channel.cpp

@@ -25,6 +25,7 @@
 #include "cloud/config.h"
 #include "load/channel/tablets_channel.h"
 #include "load/delta_writer/delta_writer.h"
+#include "runtime/thread_context.h"
 #include "storage/tablet_info.h"
 
 namespace doris {
@@ -278,7 +279,17 @@ Status CloudTabletsChannel::close(LoadChannel* parent, const PTabletWriterAddBlo
     std::vector<std::function<Status()>> tasks;
     tasks.reserve(writers_to_commit.size());
     for (auto* writer : writers_to_commit) {
-        tasks.emplace_back([writer] { return writer->commit_rowset(); });
+        tasks.emplace_back([writer, caller_bthread_id = bthread_self()] {
+            // bthread_fork_join runs inline if starting a bthread fails. The caller
+            // already has the load context attached; do not attach it twice.
+            if (bthread_self() == caller_bthread_id) {
+                return writer->commit_rowset();
+            }
+            // Empty writers initialize their rowset and bitmap token during commit.
+            // The new bthread must inherit the load's memory tracker and workload group.
+            SCOPED_ATTACH_TASK(writer->resource_context());
+            return writer->commit_rowset();
+        });
     }
     _close_status = cloud::bthread_fork_join(tasks, 10);
     if (!_close_status.ok()) {

be/src/cloud/cloud_txn_delete_bitmap_cache.cpp

@@ -26,6 +26,8 @@
 #include "cloud/config.h"
 #include "common/status.h"
 #include "cpp/sync_point.h"
+#include "runtime/thread_context.h"
+#include "runtime/workload_management/resource_context.h"
 #include "storage/olap_common.h"
 #include "storage/rowset/rowset_fwd.h"
 #include "storage/tablet/tablet_meta.h"
@@ -187,6 +189,19 @@ Status CloudTxnDeleteBitmapCache::get_delete_bitmap(
     return Status::OK();
 }
 
+std::shared_ptr<WorkloadGroup> CloudTxnDeleteBitmapCache::get_workload_group(
+        TTransactionId transaction_id, int64_t tablet_id) {
+    std::shared_lock rlock(_rwlock);
+    TxnKey key(transaction_id, tablet_id);
+    auto it = _txn_map.find(key);
+    if (it != _txn_map.end()) {
+        return it->second.workload_group;
+    }
+    auto marker = _empty_rowset_markers.find(key);
+    // A retried request on another BE has no local owner.
+    return marker == _empty_rowset_markers.end() ? nullptr : marker->second.workload_group;
+}
+
 void CloudTxnDeleteBitmapCache::set_tablet_txn_info(
         TTransactionId transaction_id, int64_t tablet_id, DeleteBitmapPtr delete_bitmap,
         const RowsetIdUnorderedSet& rowset_ids, RowsetSharedPtr rowset, int64_t txn_expiration,
@@ -204,6 +219,11 @@ void CloudTxnDeleteBitmapCache::set_tablet_txn_info(
                 std::make_shared<PublishStatus>(PublishStatus::INIT);
         _txn_map[txn_key] = TxnVal(rowset, txn_expiration, std::move(partial_update_info),
                                    std::move(publish_status), attach_row_binlog);
+        // Cloud DELETE agent tasks have no attached resource context and use the
+        // default load pool. Only capture a workload group from an attached task.
+        auto* ctx = thread_context();
+        _txn_map[txn_key].workload_group =
+                ctx->is_attach_task() ? ctx->resource_ctx()->workload_group() : nullptr;
         _expiration_txn.emplace(txn_expiration, txn_key);
     }
     std::string key_str = fmt::format("{}/{}", transaction_id, tablet_id);
@@ -304,7 +324,8 @@ void CloudTxnDeleteBitmapCache::remove_expired_tablet_txn_info() {
         }
         // Clean from _empty_rowset_markers if exists
         auto marker_iter = _empty_rowset_markers.find(iter->second);
-        if (marker_iter != _empty_rowset_markers.end()) {
+        if (marker_iter != _empty_rowset_markers.end() &&
+            iter->first == marker_iter->second.txn_expiration) {
             LOG_INFO("clean expired empty rowset marker")
                     .tag("txn_id", iter->second.txn_id)
                     .tag("tablet_id", iter->second.tablet_id);
@@ -329,6 +350,7 @@ void CloudTxnDeleteBitmapCache::remove_unused_tablet_txn_info(TTransactionId tra
         erase(cache_key);
         _txn_map.erase(txn_key);
     }
+    _empty_rowset_markers.erase(txn_key);
 }
 
 void CloudTxnDeleteBitmapCache::mark_empty_rowset(TTransactionId txn_id, int64_t tablet_id,
@@ -347,7 +369,10 @@ void CloudTxnDeleteBitmapCache::mark_empty_rowset(TTransactionId txn_id, int64_t
     }
     std::unique_lock<std::shared_mutex> wlock(_rwlock);
     TxnKey txn_key(txn_id, tablet_id);
-    _empty_rowset_markers.emplace(txn_key);
+    auto* ctx = thread_context();
+    _empty_rowset_markers[txn_key] = {
+            ctx->is_attach_task() ? ctx->resource_ctx()->workload_group() : nullptr,
+            txn_expiration};
     _expiration_txn.emplace(txn_expiration, txn_key);
 }
 
@@ -364,4 +389,4 @@ void CloudTxnDeleteBitmapCache::_clean_thread_callback() {
             std::chrono::seconds(config::remove_expired_tablet_txn_info_interval_seconds)));
 }
 
-} // namespace doris
+} // namespace doris

be/src/cloud/cloud_txn_delete_bitmap_cache.h

@@ -29,6 +29,8 @@
 
 namespace doris {
 
+class WorkloadGroup;
+
 // Record transaction related delete bitmaps using a lru cache.
 class CloudTxnDeleteBitmapCache : public LRUCachePolicy {
 public:
@@ -38,6 +40,9 @@ class CloudTxnDeleteBitmapCache : public LRUCachePolicy {
 
     Status init();
 
+    std::shared_ptr<WorkloadGroup> get_workload_group(TTransactionId transaction_id,
+                                                      int64_t tablet_id);
+
     Status get_tablet_txn_info(TTransactionId transaction_id, int64_t tablet_id,
                                RowsetSharedPtr* rowset, DeleteBitmapPtr* delete_bitmap,
                                RowsetIdUnorderedSet* rowset_ids, int64_t* txn_expiration,
@@ -68,7 +73,7 @@ class CloudTxnDeleteBitmapCache : public LRUCachePolicy {
     // Check if this is a known empty/skipped rowset
     // Returns true if was marked as empty rowset
     // Note: Does not remove the marker, as CalcDeleteBitmapTask may retry.
-    // Cleanup is handled by expiration-based removal in remove_expired_tablet_txn_info()
+    // Cleanup is handled by remove_expired_tablet_txn_info() or remove_unused_tablet_txn_info().
     bool is_empty_rowset(TTransactionId txn_id, int64_t tablet_id);
 
     // !!!ATTENTION!!!: the delete bitmap stored in CloudTxnDeleteBitmapCache contains sentinel marks,
@@ -105,6 +110,8 @@ class CloudTxnDeleteBitmapCache : public LRUCachePolicy {
     };
 
     struct TxnVal {
+        // Preserve write-stage resource isolation through commit/retries.
+        std::shared_ptr<WorkloadGroup> workload_group;
         RowsetSharedPtr rowset;
         int64_t txn_expiration;
         std::shared_ptr<PartialUpdateInfo> partial_update_info;
@@ -124,11 +131,16 @@ class CloudTxnDeleteBitmapCache : public LRUCachePolicy {
                   attach_row_binlog(attach_row_binlog_) {}
     };
 
+    struct EmptyRowsetMarker {
+        std::shared_ptr<WorkloadGroup> workload_group;
+        int64_t txn_expiration;
+    };
+
     std::map<TxnKey, TxnVal> _txn_map;
     std::multimap<int64_t, TxnKey> _expiration_txn;
-    // Lightweight markers for empty/skipped rowsets (only stores TxnKey, ~16 bytes per entry)
-    // Used to track empty rowsets that were not committed to meta-service
-    std::set<TxnKey> _empty_rowset_markers;
+    // Empty/skipped rowsets have no metadata, but their publish tasks still need
+    // the write-stage resource domain until the marker is removed or expires.
+    std::map<TxnKey, EmptyRowsetMarker> _empty_rowset_markers;
     std::shared_mutex _rwlock;
     std::shared_ptr<Thread> _clean_thread;
     CountDownLatch _stop_latch;