github-actions[bot] commented on code in PR #67761:
URL: https://github.com/apache/doris/pull/67761#discussion_r4023453675
##########
cloud/src/meta-service/meta_service_txn.cpp:
##########
@@ -4653,75 +4676,116 @@ void MetaServiceImpl::get_prepare_txn_by_coordinator(
}
std::unique_ptr<RangeGetIterator> it;
int32_t result_count = 0;
- int64_t total_iteration_cnt = 0;
+ int64_t scanned_count = 0;
bool has_start_time_filter = request->has_start_time();
- do {
- err = txn->get(begin_info_key, end_info_key, &it, true);
- if (err != TxnErrorCode::TXN_OK) {
- code = cast_as<ErrCategory::READ>(err);
- ss << "failed to get txn info. err=" << err;
- msg = ss.str();
+ auto process_txn_info = [&](std::string_view key, std::string_view value)
-> TxnErrorCode {
+ scanned_count++;
+ VLOG_DEBUG << "check txn info txn_info_key=" << hex(key);
+ TxnInfoPB info_pb;
+ if (!info_pb.ParseFromArray(value.data(), value.size())) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = "malformed txn info, key=" + hex(key);
LOG(WARNING) << msg;
- return;
+ return TxnErrorCode::TXN_INVALID_DATA;
+ }
+ const auto& coordinate = info_pb.coordinator();
+ bool matches = info_pb.status() == TxnStatusPB::TXN_STATUS_PREPARED &&
+ coordinate.sourcetype() == TXN_SOURCE_TYPE_BE &&
+ coordinate.ip() == request->ip() &&
+ (coordinate.id() == 0 || coordinate.id() ==
request->id());
+ if (matches && has_start_time_filter) {
+ matches = coordinate.start_time() < request->start_time();
+ }
+ if (matches) {
+ TxnInfoPB* txn_info = response->add_txn_infos();
+ txn_info->CopyFrom(info_pb);
+ result_count++;
}
+ return TxnErrorCode::TXN_OK;
+ };
+ // Each txn_info value can be much larger than its running index entry.
+ constexpr int batch_size = 128;
+ const int scan_batch_size = scan_by_running_key ? batch_size :
RangeGetOptions().batch_limit;
+ auto read_page = [&]() -> TxnErrorCode {
+ auto ret = txn->get(begin_key, end_key, &it, true, scan_batch_size);
+ TEST_SYNC_POINT_CALLBACK("get_prepare_txn_by_coordinator::range_get",
&ret);
+ if (ret != TxnErrorCode::TXN_OK) {
+ return ret;
+ }
+ std::vector<std::string> info_keys;
+ info_keys.reserve(scan_by_running_key ? it->size() : 0);
while (it->has_next()) {
- total_iteration_cnt++;
- auto [k, v] = it->next();
- VLOG_DEBUG << "check txn info txn_info_key=" << hex(k);
- TxnInfoPB info_pb;
- if (!info_pb.ParseFromArray(v.data(), v.size())) {
- code = MetaServiceCode::PROTOBUF_PARSE_ERR;
- ss << "malformed txn running info";
- msg = ss.str();
- ss << " key=" << hex(k);
- LOG(WARNING) << ss.str();
- return;
+ auto [key, value] = it->next();
+ if (scan_by_running_key) {
+ auto info_key = get_txn_info_key_from_txn_running_key(key);
+ if (info_key.empty()) {
+ continue;
+ }
+ info_keys.push_back(std::move(info_key));
+ } else {
+ ret = process_txn_info(key, value);
+ if (ret != TxnErrorCode::TXN_OK) {
+ return ret;
+ }
}
- const auto& coordinate = info_pb.coordinator();
- bool matches = info_pb.status() ==
TxnStatusPB::TXN_STATUS_PREPARED &&
- coordinate.sourcetype() == TXN_SOURCE_TYPE_BE &&
- coordinate.ip() == request->ip() &&
- (coordinate.id() == 0 || coordinate.id() ==
request->id());
- if (matches && has_start_time_filter) {
- matches = coordinate.start_time() < request->start_time();
+ }
+ if (!scan_by_running_key) {
+ return TxnErrorCode::TXN_OK;
+ }
+ std::vector<std::optional<std::string>> info_values;
+ ret = txn->batch_get(&info_values, info_keys,
Transaction::BatchGetOptions(true));
+ TEST_SYNC_POINT_CALLBACK("get_prepare_txn_by_coordinator::batch_get",
&ret, &info_values);
+ if (ret != TxnErrorCode::TXN_OK) {
+ return ret;
+ }
+ for (size_t i = 0; i < info_keys.size(); ++i) {
+ if (!info_values[i].has_value()) {
+ code = MetaServiceCode::TXN_ID_NOT_FOUND;
+ msg = "missing txn info for running txn, key=" +
hex(info_keys[i]);
+ LOG(WARNING) << msg;
+ return TxnErrorCode::TXN_KEY_NOT_FOUND;
}
-
- if (matches) {
- TxnInfoPB* txn_info = response->add_txn_infos();
- txn_info->CopyFrom(info_pb);
- result_count++;
+ ret = process_txn_info(info_keys[i], *info_values[i]);
+ if (ret != TxnErrorCode::TXN_OK) {
+ return ret;
}
+ }
+ return TxnErrorCode::TXN_OK;
+ };
- if (!it->has_next()) {
- begin_info_key = k;
+ do {
+ err = read_page();
+ if (err == TxnErrorCode::TXN_TOO_OLD) {
Review Comment:
[P2] Destroy failed range futures before retrying
A real production FDB range `TXN_TOO_OLD` returns from
`Transaction::get(begin, end, ...)` before its raw `FDBFuture*` is transferred
to `RangeGetIterator`; `txn.reset()` here destroys only the `FDBTransaction`,
not that future. The helper's raw-pointer bug predates this PR, but this new
local retry can repeat it once per expired page (or twice when the retry also
fails) and then return `OK` while retaining those futures. The MemTxnKv
callbacks inject after a successful owned read and cannot exercise this
lifetime path. Please give the range future immediate RAII ownership, release
it only when transferring it to `RangeGetIterator`, and add an FDB-backed
error-path ownership test.
--
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]
