github-actions[bot] commented on code in PR #67761:
URL: https://github.com/apache/doris/pull/67761#discussion_r3975700383
##########
cloud/src/meta-service/meta_service_txn.cpp:
##########
@@ -4653,75 +4675,115 @@ 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;
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 {
Review Comment:
[P1] Revalidate matches after changing the read version
Once a later page returns `TXN_TOO_OLD`, this creates a new snapshot but
leaves `txn_infos` already appended from earlier pages. For example, page 1 can
append A as `PREPARED`, then `precommit_txn` changes A to `PRECOMMITTED` while
retaining its running key; this retry resumes past A, so the successful
response still contains stale A. FE discards each returned transaction's status
and aborts by db/txn ID, and the abort path permits `PRECOMMITTED`, so
coordinator cleanup can roll back a live 2PC transaction that this RPC's
`PREPARED` filter is meant to exclude. Before this change, the handler returned
`TXN_TOO_OLD`; the default outer proxy retry cleared and rescanned the whole
response. Please revalidate accumulated matches in the renewed view or
restart/clear the scan, and cover an earlier-page `PREPARED`-to-`PRECOMMITTED`
transition in the retry 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]
