[feature](cloud) Add lazy commit mechanism for commit_txn

be/src/cloud/cloud_meta_mgr.cpp

@@ -847,6 +847,7 @@ Status CloudMetaMgr::commit_txn(const StreamLoadContext& ctx, bool is_2pc) {
     req.set_db_id(ctx.db_id);
     req.set_txn_id(ctx.txn_id);
     req.set_is_2pc(is_2pc);
+    req.set_enable_txn_lazy_commit(config::enable_cloud_txn_lazy_commit);
     auto st = retry_rpc("commit txn", req, &res, &MetaService_Stub::commit_txn);
 
     if (st.ok()) {

be/src/cloud/config.cpp

@@ -62,4 +62,6 @@ DEFINE_mInt32(sync_load_for_tablets_thread, "32");
 
 DEFINE_mBool(enable_new_tablet_do_compaction, "false");
 
+DEFINE_Bool(enable_cloud_txn_lazy_commit, "false");
+
 } // namespace doris::config

be/src/cloud/config.h

@@ -95,4 +95,7 @@ DECLARE_mBool(save_load_error_log_to_s3);
 // the theads which sync the datas which loaded in other clusters
 DECLARE_mInt32(sync_load_for_tablets_thread);
 
+// enable large txn lazy commit in meta-service `commit_txn`
+DECLARE_mBool(enable_cloud_txn_lazy_commit);
+
 } // namespace doris::config

cloud/src/common/config.h

@@ -206,4 +206,11 @@ CONF_Validator(s3_client_http_scheme, [](const std::string& config) -> bool {
 // Max retry times for object storage request
 CONF_mInt64(max_s3_client_retry, "10");
 
+CONF_Bool(enable_cloud_txn_lazy_commit, "true");
+CONF_Int32(txn_lazy_commit_rowsets_thresold, "1000");
+CONF_Int32(txn_lazy_commit_num_threads, "8");
+CONF_Int32(txn_lazy_max_rowsets_per_batch, "1000");
+
+// max TabletIndexPB num for batch get
+CONF_Int32(max_tablet_index_num_per_batch, "1000");
 } // namespace doris::cloud::config

cloud/src/meta-service/CMakeLists.txt

@@ -24,4 +24,5 @@ add_library(MetaService
     doris_txn.cpp
     mem_txn_kv.cpp
     http_encode_key.cpp
+    txn_lazy_committer.cpp
 )

cloud/src/meta-service/meta_service.h

@@ -30,6 +30,7 @@
 #include "common/config.h"
 #include "cpp/sync_point.h"
 #include "meta-service/txn_kv.h"
+#include "meta-service/txn_lazy_committer.h"
 #include "rate-limiter/rate_limiter.h"
 #include "resource-manager/resource_manager.h"
 
@@ -296,6 +297,7 @@ class MetaServiceImpl : public cloud::MetaService {
     std::shared_ptr<TxnKv> txn_kv_;
     std::shared_ptr<ResourceManager> resource_mgr_;
     std::shared_ptr<RateLimiter> rate_limiter_;
+    std::shared_ptr<TxnLazyCommitter> txn_lazy_committer_;
 };
 
 class MetaServiceProxy final : public MetaService {