gavinchou commented on code in PR #67820:
URL: https://github.com/apache/doris/pull/67820#discussion_r3991200982
##########
cloud/src/meta-service/meta_service_txn.cpp:
##########
@@ -4724,6 +4772,171 @@ std::string
get_txn_info_key_from_txn_running_key(std::string_view txn_running_k
return conflict_txn_info_key;
}
+void MetaServiceImpl::advance_tso_fence(::google::protobuf::RpcController*
controller,
+ const AdvanceTsoFenceRequest* request,
+ AdvanceTsoFenceResponse* response,
+ ::google::protobuf::Closure* done) {
+ RPC_PREPROCESS(advance_tso_fence, get, put);
+ if (!request->has_proposed_fence_tso() || request->proposed_fence_tso() <=
0) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "invalid proposed TSO fence";
+ return;
+ }
+ instance_id = get_instance_id(resource_mgr_, request->cloud_unique_id());
+ if (instance_id.empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "cannot find instance_id for TSO fence";
+ return;
+ }
+ RPC_RATE_LIMIT(advance_tso_fence)
+
+ TxnErrorCode err = txn_kv_->create_txn(&txn);
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::CREATE>(err);
+ msg = "failed to create TSO fence transaction";
+ return;
+ }
+
+ const std::string key = txn_tso_fence_key({instance_id});
+ std::string value;
+ err = txn->get(key, &value);
+ int64_t current_fence_tso = 0;
+ if (err == TxnErrorCode::TXN_OK) {
+ TxnTsoFencePB fence;
+ if (!fence.ParseFromString(value) || !fence.has_fence_tso() ||
fence.fence_tso() <= 0) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = "failed to parse TSO fence";
+ return;
+ }
+ current_fence_tso = fence.fence_tso();
+ } else if (err != TxnErrorCode::TXN_KEY_NOT_FOUND) {
+ code = cast_as<ErrCategory::READ>(err);
+ msg = "failed to read TSO fence";
+ return;
+ }
+
+ const int64_t effective_fence_tso = std::max(current_fence_tso,
request->proposed_fence_tso());
+ if (effective_fence_tso > current_fence_tso) {
+ TxnTsoFencePB fence;
+ fence.set_fence_tso(effective_fence_tso);
+ if (!fence.SerializeToString(&value)) {
+ code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR;
+ msg = "failed to serialize TSO fence";
+ return;
+ }
+ txn->put(key, value);
+ err = txn->commit();
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::COMMIT>(err);
+ msg = "failed to commit TSO fence";
+ return;
+ }
+ }
+ response->set_tso_fence(effective_fence_tso);
+}
+
+void MetaServiceImpl::get_tso_recovery_transactions(
+ ::google::protobuf::RpcController* controller,
+ const GetTsoRecoveryTransactionsRequest* request,
+ GetTsoRecoveryTransactionsResponse* response,
::google::protobuf::Closure* done) {
+ RPC_PREPROCESS(get_tso_recovery_transactions, get);
+ if (request->end_txn_id() <= 0 || request->batch_size() <= 0 ||
request->batch_size() > 1000 ||
+ request->tso_fence() <= 0) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "invalid TSO recovery transaction bound or batch size";
+ return;
+ }
+ instance_id = get_instance_id(resource_mgr_, request->cloud_unique_id());
+ if (instance_id.empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "cannot find instance_id for TSO recovery";
+ return;
+ }
+ RPC_RATE_LIMIT(get_tso_recovery_transactions)
+ // Keys sort by database first. Scan the instance and apply the fixed
exclusive ID bound
+ // to each key; a batch containing only newer transactions does not finish
the scan.
+ std::string begin_key = txn_running_key({instance_id, 0, 0});
+ std::string end_key = txn_running_key({instance_id, INT64_MAX, INT64_MAX});
+ end_key.push_back('\x00');
+ if (!request->start_key().empty()) {
+ if (request->start_key() < begin_key || request->start_key() >=
end_key) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "invalid TSO recovery start key";
+ return;
+ }
+ begin_key = request->start_key();
+ }
+ TxnErrorCode err = txn_kv_->create_txn(&txn);
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::CREATE>(err);
+ msg = "failed to create TSO recovery read transaction";
+ return;
+ }
+ std::unique_ptr<RangeGetIterator> it;
+ err = txn->get(begin_key, end_key, &it, true, request->batch_size());
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::READ>(err);
+ msg = "failed to get running transactions during TSO recovery";
+ return;
+ }
+ while (it->has_next()) {
Review Comment:
this `while` loop may exceed 5 seconds limit of FDB txn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
