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

Motivation:
* In cloud mode we use `foundationdb` for storing rowset meta, when we load data with many rowsets,
  `commit_txn` will failed because `foundationdb` currently limits all transactions to be below 10 MB in size,
  So we spilt `commit_txn` into several sub fdb txn to solve the problem

How:

`commit_txn` flow like this

sub txn 1:
  1. update partition `VersionPB` with txn_id
  2. submit async task to txn_lazy_committer

txn_lazy_committer:
  sub txn 2:
      convert tmp rowset meta per batch
  sub txn 3:
      convert tmp rowset meta per batch
  ......
  sub txn n:
      make txn visible and remove txn_id in `VersionPB`

`get_version` main flow like this:

  1. `VersionPB` without txn_id, the same as before

  2. `VersionPB` has txn_id field, it means this is a txn
      need be advanced by txn_lazy_committer, so submit a
      `txn_lazy_task` to make last txn visible and retry

Author: SWJTU-ZhangLei

cloud/src/common/config.h

@@ -201,4 +201,8 @@ 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_txn_lazy_commit, "true");
+CONF_Int32(txn_lazy_commit_rowsets_thresold, "2");
+CONF_Int32(txn_lazy_commit_worker_num, "8");
+CONF_Int32(txn_lazy_max_rowsets_per_batch, "256");
 } // 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.cpp

@@ -75,6 +75,7 @@ MetaServiceImpl::MetaServiceImpl(std::shared_ptr<TxnKv> txn_kv,
     resource_mgr_ = resource_mgr;
     rate_limiter_ = rate_limiter;
     rate_limiter_->init(this);
+    txn_lazy_committer_ = std::make_shared<TxnLazyCommitter>(txn_kv_);
 }
 
 MetaServiceImpl::~MetaServiceImpl() = default;
@@ -237,46 +238,64 @@ void MetaServiceImpl::get_version(::google::protobuf::RpcController* controller,
         partition_version_key({instance_id, db_id, table_id, partition_id}, &ver_key);
     }
 
-    std::unique_ptr<Transaction> txn;
-    TxnErrorCode err = txn_kv_->create_txn(&txn);
-    if (err != TxnErrorCode::TXN_OK) {
-        msg = "failed to create txn";
-        code = cast_as<ErrCategory::CREATE>(err);
-        return;
-    }
+    do {
+        code = MetaServiceCode::OK;
+        std::unique_ptr<Transaction> txn;
+        TxnErrorCode err = txn_kv_->create_txn(&txn);
+        if (err != TxnErrorCode::TXN_OK) {
+            msg = "failed to create txn";
+            code = cast_as<ErrCategory::CREATE>(err);
+            return;
+        }
 
-    std::string ver_val;
-    // 0 for success get a key, 1 for key not found, negative for error
-    err = txn->get(ver_key, &ver_val);
-    VLOG_DEBUG << "xxx get version_key=" << hex(ver_key);
-    if (err == TxnErrorCode::TXN_OK) {
-        if (is_table_version) {
-            int64_t version = 0;
-            if (!txn->decode_atomic_int(ver_val, &version)) {
-                code = MetaServiceCode::PROTOBUF_PARSE_ERR;
-                msg = "malformed table version value";
-                return;
-            }
-            response->set_version(version);
-        } else {
-            VersionPB version_pb;
-            if (!version_pb.ParseFromString(ver_val)) {
-                code = MetaServiceCode::PROTOBUF_PARSE_ERR;
-                msg = "malformed version value";
-                return;
+        std::string ver_val;
+        // 0 for success get a key, 1 for key not found, negative for error
+        err = txn->get(ver_key, &ver_val);
+        VLOG_DEBUG << "xxx get version_key=" << hex(ver_key);
+        if (err == TxnErrorCode::TXN_OK) {
+            if (is_table_version) {
+                int64_t version = 0;
+                if (!txn->decode_atomic_int(ver_val, &version)) {
+                    code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+                    msg = "malformed table version value";
+                    return;
+                }
+                response->set_version(version);
+            } else {
+                VersionPB version_pb;
+                if (!version_pb.ParseFromString(ver_val)) {
+                    code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+                    msg = "malformed version value";
+                    return;
+                }
+
+                if (version_pb.has_txn_id()) {
+                    txn.reset();
+                    std::shared_ptr<TxnLazyCommitTask> task =
+                            txn_lazy_committer_->submit(instance_id, version_pb.txn_id());
+                    std::pair<MetaServiceCode, std::string> ret = task->wait();
+                    code = ret.first;
+                    msg = ret.second;
+                    if (code != MetaServiceCode::OK) {
+                        LOG(WARNING)
+                                << "wait txn lazy commit failed, txn_id=" << version_pb.txn_id();
+                        return;
+                    }
+                }
+
+                response->set_version(version_pb.version());
+                response->add_version_update_time_ms(version_pb.update_time_ms());
             }
-            response->set_version(version_pb.version());
-            response->add_version_update_time_ms(version_pb.update_time_ms());
+            { TEST_SYNC_POINT_CALLBACK("get_version_code", &code); }
+            return;
+        } else if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
+            msg = "not found";
+            code = MetaServiceCode::VERSION_NOT_FOUND;
+            return;
         }
-        { TEST_SYNC_POINT_CALLBACK("get_version_code", &code); }
-        return;
-    } else if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
-        msg = "not found";
-        code = MetaServiceCode::VERSION_NOT_FOUND;
-        return;
-    }
-    msg = fmt::format("failed to get txn, err={}", err);
-    code = cast_as<ErrCategory::READ>(err);
+        msg = fmt::format("failed to get txn, err={}", err);
+        code = cast_as<ErrCategory::READ>(err);
+    } while (false);
 }
 
 void MetaServiceImpl::batch_get_version(::google::protobuf::RpcController* controller,
@@ -326,8 +345,13 @@ void MetaServiceImpl::batch_get_version(::google::protobuf::RpcController* contr
     std::vector<std::optional<std::string>> version_values;
     version_keys.reserve(BATCH_SIZE);
     version_values.reserve(BATCH_SIZE);
+
     while ((code == MetaServiceCode::OK || code == MetaServiceCode::KV_TXN_TOO_OLD) &&
            response->versions_size() < response->partition_ids_size()) {
+    TRY_AGAIN:
+        response->clear_versions();
+        code = MetaServiceCode::OK;
+
         std::unique_ptr<Transaction> txn;
         TxnErrorCode err = txn_kv_->create_txn(&txn);
         if (err != TxnErrorCode::TXN_OK) {
@@ -387,10 +411,30 @@ void MetaServiceImpl::batch_get_version(::google::protobuf::RpcController* contr
                         msg = "malformed version value";
                         break;
                     }
+                    if (version_pb.has_txn_id()) {
+                        txn.reset();
+                        std::shared_ptr<TxnLazyCommitTask> task =
+                                txn_lazy_committer_->submit(instance_id, version_pb.txn_id());
+                        std::pair<MetaServiceCode, std::string> ret = task->wait();
+                        code = ret.first;
+                        msg = ret.second;
+                        if (code != MetaServiceCode::OK) {
+                            LOG(WARNING) << "wait txn lazy commit failed, txn_id="
+                                         << version_pb.txn_id();
+                            break;
+                        }
+                        goto TRY_AGAIN;
+                    }
                     response->add_versions(version_pb.version());
                     response->add_version_update_time_ms(version_pb.update_time_ms());
                 }
             }
+            if (code != MetaServiceCode::OK) {
+                break;
+            }
+        }
+        if (code != MetaServiceCode::OK) {
+            break;
         }
     }
     if (code != MetaServiceCode::OK) {

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"
 
@@ -292,6 +293,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 {