luwei16 commented on code in PR #67820:
URL: https://github.com/apache/doris/pull/67820#discussion_r4012180041
##########
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()) {
+ auto [key, value] = it->next();
+ if (!it->has_next()) {
+ begin_key = key;
+ }
+ std::string_view encoded_key = key;
+ encoded_key.remove_prefix(1);
+ std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>>
fields;
+ if (decode_key(&encoded_key, &fields) != 0 || fields.size() != 5 ||
+ !std::holds_alternative<int64_t>(std::get<0>(fields[3])) ||
+ !std::holds_alternative<int64_t>(std::get<0>(fields[4]))) {
+ code = MetaServiceCode::UNDEFINED_ERR;
+ msg = "failed to decode running transaction key during TSO
recovery";
+ return;
+ }
+ const auto db_id = std::get<int64_t>(std::get<0>(fields[3]));
+ const auto txn_id = std::get<int64_t>(std::get<0>(fields[4]));
+ if (txn_id >= request->end_txn_id()) {
+ continue;
+ }
+ // Running keys are removed atomically with real VISIBLE/ABORTED.
Expired COMMITTED
+ // lazy transactions still block. Read the details in the same KV
snapshot.
+ std::string info_val;
+ err = txn->get(txn_info_key({instance_id, db_id, txn_id}), &info_val,
true);
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::READ>(err);
+ msg = fmt::format("failed to read TSO recovery transaction,
db_id={}, txn_id={}", db_id,
+ txn_id);
+ return;
+ }
+ TxnInfoPB info;
+ if (!info.ParseFromString(info_val)) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = "failed to parse TSO recovery transaction";
+ return;
+ }
+ // Subtransactions update txn_info.table_ids; the running record can
be older.
+ if (info.db_id() != db_id || info.txn_id() != txn_id ||
info.table_ids().empty()) {
+ code = MetaServiceCode::UNDEFINED_ERR;
+ msg = "invalid TSO recovery transaction identity or tables";
+ return;
+ }
+ if (info.status() != TxnStatusPB::TXN_STATUS_COMMITTED ||
!info.has_commit_tso() ||
+ info.commit_tso() <= 0 || info.commit_tso() >
request->tso_fence()) {
+ continue;
+ }
+ // Recovery needs identities and visibility boundaries, not commit
attachments.
+ auto* recovered = response->add_txn_infos();
+ recovered->set_db_id(db_id);
+ recovered->set_txn_id(txn_id);
+ recovered->mutable_table_ids()->CopyFrom(info.table_ids());
+ recovered->set_status(info.status());
+ recovered->set_commit_tso(info.commit_tso());
+ }
+ begin_key.push_back('\x00');
+ response->set_next_start_key(it->more() ? begin_key : "");
Review Comment:
The recovery RPC and raw-key transfer have been removed.
##########
gensrc/proto/cloud.proto:
##########
@@ -1256,6 +1263,38 @@ message CheckTxnConflictResponse {
repeated TxnInfoPB conflict_txns = 3;
}
+message GetTsoRecoveryTransactionsRequest {
+ optional string cloud_unique_id = 1; // For auth
+ // Exclusive transaction-ID bound, fixed for the complete recovery scan.
+ optional int64 end_txn_id = 2;
+ // Maximum running records scanned per batch, including IDs above the
bound.
+ optional int32 batch_size = 3;
+ optional bytes start_key = 4;
+ optional string request_ip = 5;
+ // Only committed TSO transactions at or below this inclusive fence are
recovered.
+ optional int64 tso_fence = 6;
+}
+
+message GetTsoRecoveryTransactionsResponse {
+ optional MetaServiceResponseStatus status = 1;
+ // Includes expired running transactions from every database/table in the
instance.
+ repeated TxnInfoPB txn_infos = 2;
+ // Always present on success; empty means the full instance has been
scanned.
+ optional bytes next_start_key = 3;
Review Comment:
The raw-key recovery field has been removed.
##########
gensrc/proto/cloud.proto:
##########
@@ -1256,6 +1263,38 @@ message CheckTxnConflictResponse {
repeated TxnInfoPB conflict_txns = 3;
}
+message GetTsoRecoveryTransactionsRequest {
+ optional string cloud_unique_id = 1; // For auth
+ // Exclusive transaction-ID bound, fixed for the complete recovery scan.
+ optional int64 end_txn_id = 2;
Review Comment:
The recovery flow no longer uses a transaction-ID watermark or assumes
monotonic transaction IDs.
##########
gensrc/proto/cloud.proto:
##########
@@ -1256,6 +1263,38 @@ message CheckTxnConflictResponse {
repeated TxnInfoPB conflict_txns = 3;
}
+message GetTsoRecoveryTransactionsRequest {
+ optional string cloud_unique_id = 1; // For auth
+ // Exclusive transaction-ID bound, fixed for the complete recovery scan.
+ optional int64 end_txn_id = 2;
+ // Maximum running records scanned per batch, including IDs above the
bound.
+ optional int32 batch_size = 3;
+ optional bytes start_key = 4;
Review Comment:
The recovery pagination API and `start_key` have been removed.
##########
gensrc/proto/cloud.proto:
##########
@@ -1256,6 +1263,38 @@ message CheckTxnConflictResponse {
repeated TxnInfoPB conflict_txns = 3;
}
+message GetTsoRecoveryTransactionsRequest {
+ optional string cloud_unique_id = 1; // For auth
+ // Exclusive transaction-ID bound, fixed for the complete recovery scan.
+ optional int64 end_txn_id = 2;
+ // Maximum running records scanned per batch, including IDs above the
bound.
+ optional int32 batch_size = 3;
+ optional bytes start_key = 4;
+ optional string request_ip = 5;
+ // Only committed TSO transactions at or below this inclusive fence are
recovered.
+ optional int64 tso_fence = 6;
+}
+
+message GetTsoRecoveryTransactionsResponse {
+ optional MetaServiceResponseStatus status = 1;
+ // Includes expired running transactions from every database/table in the
instance.
+ repeated TxnInfoPB txn_infos = 2;
+ // Always present on success; empty means the full instance has been
scanned.
+ optional bytes next_start_key = 3;
+}
+
+message AdvanceTsoFenceRequest {
+ optional string cloud_unique_id = 1; // For auth
+ // Inclusive upper bound rejected by transaction commit.
+ optional int64 proposed_fence_tso = 2;
+ optional string request_ip = 3;
Review Comment:
Kept `request_ip`, consistent with the existing Meta Service RPC fields.
##########
cloud/src/meta-store/keys.cpp:
##########
@@ -259,6 +261,11 @@ void txn_running_key(const TxnRunningKeyInfo& in,
std::string* out) {
encode_int64(std::get<2>(in), out); // txn_id
}
+void txn_tso_fence_key(const TxnTsoFenceKeyInfo& in, std::string* out) {
Review Comment:
Added `TxnTsoFenceKey` support to the Meta Service HTTP encode/get/set paths
and tests.
##########
fe/fe-core/src/main/java/org/apache/doris/tso/TSOServiceState.java:
##########
@@ -0,0 +1,66 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.tso;
+
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.google.gson.annotations.SerializedName;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/** The durable allocation window and readable transaction prefix, published
as one snapshot. */
+public final class TSOServiceState implements Writable {
+ // Keep the old TSOTimestamp JSON fields and image checksum for
journal/image compatibility.
+ @SerializedName("physicalTimestamp")
+ private final long windowEndPhysicalTime;
Review Comment:
Removed the compatibility fields and renamed the field/getter with the `Ms`
unit.
--
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]
