mymeiyi commented on code in PR #67761:
URL: https://github.com/apache/doris/pull/67761#discussion_r3977280974
##########
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;
Review Comment:
Added `reserve()` for `info_keys`. `batch_get()` already calls `reserve()`
internally for info_values.
--
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]
