luwei16 commented on code in PR #67820:
URL: https://github.com/apache/doris/pull/67820#discussion_r4012179243
##########
gensrc/proto/cloud.proto:
##########
@@ -1949,6 +1988,8 @@ enum MetaServiceCode {
STALE_TABLET_CACHE = 2012;
STALE_PREPARE_ROWSET = 2013;
TXN_ALREADY_COMMITED = 2014;
+ // The transaction commit TSO belongs to an earlier FE master epoch.
+ TXN_COMMIT_TSO_FENCED = 2015;
Review Comment:
Renamed it to `TXN_COMMIT_TSO_EXPIRED`.
##########
cloud/src/meta-service/meta_service_txn.cpp:
##########
@@ -127,6 +127,39 @@ static void
append_table_stream_commit_size_error(TxnErrorCode err, std::string&
}
}
+static bool check_txn_commit_tso_fence(Transaction* txn, const std::string&
instance_id,
Review Comment:
Added a comment describing the fence read, conflict behavior, and
allow/reject cases.
##########
cloud/src/meta-service/meta_service_txn.cpp:
##########
@@ -1911,6 +1944,11 @@ void MetaServiceImpl::commit_txn_immediately(
return;
}
+ if (txn_info.status() != TxnStatusPB::TXN_STATUS_COMMITTED &&
Review Comment:
Added both the request flag and mutable Meta Service config checks.
##########
cloud/src/meta-service/meta_service_txn.cpp:
##########
@@ -2743,6 +2781,11 @@ void MetaServiceImpl::commit_txn_eventually(
return;
}
+ if (txn_info.status() != TxnStatusPB::TXN_STATUS_COMMITTED &&
Review Comment:
Added the same two checks to the eventual commit path.
##########
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) {
Review Comment:
FE treats this as a hard failure: startup remains uninitialized; an
uncertain-commit fence failure deactivates TSO and keeps the read prefix
unavailable.
##########
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:
The recovery scan and this loop have been removed in the latest design.
--
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]
