From 038d77a3eb6732308fa3f874b2df532983d9933a Mon Sep 17 00:00:00 2001 From: meiyi Date: Wed, 9 Sep 2026 19:07:14 +0800 Subject: [PATCH 1/2] [fix](cloud) Bound and resume coordinator transaction scans ### What problem does this PR solve? Scan running keys and batch-read txn info in groups of 128. Process each page locally with one iterator loop and resume expired pages without losing results or KV read metrics. The dynamic enable_get_prepare_txn_by_coordinator_by_running_key switch defaults to true; false scans txn info directly. Skip undecodable running keys with a warning. Log scanned and matched counts and identify the RPC in scan errors. ### Release note Reduce coordinator cleanup scan cost, recover from KV transaction expiry, and allow switching to direct txn-info scans. ### Check List (For Author) - Test: Cloud ASAN unit-test build and all 5 coordinator tests passed on the development host. Coverage includes both scan modes, single and multiple page expiry, exhausted page retries and proxy restart, filtering boundaries, malformed records, empty scans, read metrics, and matching results across 128-entry batches. Formatting, git diff --check and clang-tidy passed. - Behavior changed: Yes; bounded running-key scans, expiry recovery and configurable scan strategy. - Does this need documentation: No --- cloud/src/common/config.h | 4 + cloud/src/meta-service/meta_service_txn.cpp | 178 +++++--- cloud/src/meta-store/mem_txn_kv.cpp | 2 - cloud/test/meta_service_test.cpp | 448 ++++++++++++++++++++ 4 files changed, 572 insertions(+), 60 deletions(-) diff --git a/cloud/src/common/config.h b/cloud/src/common/config.h index faae59ad7e33f2..45bd1d836adc3d 100644 --- a/cloud/src/common/config.h +++ b/cloud/src/common/config.h @@ -353,6 +353,10 @@ CONF_Bool(delete_bitmap_enable_retry_txn_conflict, "true"); // more reasonable. CONF_mInt64(max_txn_commit_byte, "7340032"); +// true: scan txn_running_key entries and fetch transaction info via the corresponding txn_info_key. +// false: scan txn_info_key entries directly; these usually far outnumber txn_running_key entries. +CONF_mBool(enable_get_prepare_txn_by_coordinator_by_running_key, "true"); + CONF_Bool(enable_cloud_txn_lazy_commit, "true"); CONF_Int32(txn_lazy_commit_rowsets_thresold, "1000"); CONF_Int32(txn_lazy_commit_num_threads, "8"); diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index 76963cb0627e8f..135a2c38837f2a 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/cloud/src/meta-service/meta_service_txn.cpp @@ -4620,6 +4620,25 @@ void MetaServiceImpl::abort_txn_with_coordinator(::google::protobuf::RpcControll } } +std::string get_txn_info_key_from_txn_running_key(std::string_view txn_running_key) { + std::string conflict_txn_info_key; + std::vector, int, int>> out; + txn_running_key.remove_prefix(1); + int ret = decode_key(&txn_running_key, &out); + if (ret != 0) [[unlikely]] { + // decode version key error means this is something wrong, + // we can not continue this txn + LOG(WARNING) << "failed to decode key, ret=" << ret << " key=" << hex(txn_running_key); + } else { + DCHECK(out.size() == 5) << " key=" << hex(txn_running_key) << " " << out.size(); + const std::string& decode_instance_id = std::get<1>(std::get<0>(out[1])); + int64_t db_id = std::get<0>(std::get<0>(out[3])); + int64_t txn_id = std::get<0>(std::get<0>(out[4])); + conflict_txn_info_key = txn_info_key({decode_instance_id, db_id, txn_id}); + } + return conflict_txn_info_key; +} + void MetaServiceImpl::get_prepare_txn_by_coordinator( ::google::protobuf::RpcController* controller, const GetPrepareTxnByCoordinatorRequest* request, @@ -4641,9 +4660,12 @@ void MetaServiceImpl::get_prepare_txn_by_coordinator( return; } RPC_RATE_LIMIT(get_prepare_txn_by_coordinator); - std::string begin_info_key = txn_info_key({instance_id, 0, 0}); - std::string end_info_key = txn_info_key({instance_id, INT64_MAX, INT64_MAX}); - LOG(INFO) << "begin_info_key:" << hex(begin_info_key) << " end_info_key:" << hex(end_info_key); + const bool scan_by_running_key = config::enable_get_prepare_txn_by_coordinator_by_running_key; + std::string begin_key = scan_by_running_key ? txn_running_key({instance_id, 0, 0}) + : txn_info_key({instance_id, 0, 0}); + std::string end_key = scan_by_running_key ? txn_running_key({instance_id, INT64_MAX, INT64_MAX}) + : txn_info_key({instance_id, INT64_MAX, INT64_MAX}); + LOG(INFO) << "begin_key:" << hex(begin_key) << " end_key:" << hex(end_key); TxnErrorCode err = txn_kv_->create_txn(&txn); if (err != TxnErrorCode::TXN_OK) { @@ -4653,75 +4675,115 @@ void MetaServiceImpl::get_prepare_txn_by_coordinator( } std::unique_ptr 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(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 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> 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) { + stats.get_bytes += txn->get_bytes(); + stats.get_counter += txn->num_get_keys(); + txn.reset(); + err = txn_kv_->create_txn(&txn); + if (err != TxnErrorCode::TXN_OK) { + msg = "failed to create txn"; + code = cast_as(err); + return; } + err = read_page(); + } + if (code != MetaServiceCode::OK) { + return; + } + if (err != TxnErrorCode::TXN_OK) { + code = cast_as(err); + ss << "get_prepare_txn_by_coordinator: failed to get txn info. err=" << err; + msg = ss.str(); + LOG(WARNING) << msg; + return; } - begin_info_key.push_back('\x00'); // Update to next smallest key for iteration - } while (it->more()); - LOG(INFO) << "get_prepare_txn_by_coordinator: found " << result_count << " transactions" - << " total iteration count: " << total_iteration_cnt; -} + begin_key = it->next_begin_key(); + } while (it->more()); -std::string get_txn_info_key_from_txn_running_key(std::string_view txn_running_key) { - std::string conflict_txn_info_key; - std::vector, int, int>> out; - txn_running_key.remove_prefix(1); - int ret = decode_key(&txn_running_key, &out); - if (ret != 0) [[unlikely]] { - // decode version key error means this is something wrong, - // we can not continue this txn - LOG(WARNING) << "failed to decode key, ret=" << ret << " key=" << hex(txn_running_key); - } else { - DCHECK(out.size() == 5) << " key=" << hex(txn_running_key) << " " << out.size(); - const std::string& decode_instance_id = std::get<1>(std::get<0>(out[1])); - int64_t db_id = std::get<0>(std::get<0>(out[3])); - int64_t txn_id = std::get<0>(std::get<0>(out[4])); - conflict_txn_info_key = txn_info_key({decode_instance_id, db_id, txn_id}); - } - return conflict_txn_info_key; + LOG(INFO) << "get_prepare_txn_by_coordinator: scanned_count=" << scanned_count + << " matched_count=" << result_count + << " scan_by_running_key=" << scan_by_running_key; } void MetaServiceImpl::check_txn_conflict(::google::protobuf::RpcController* controller, diff --git a/cloud/src/meta-store/mem_txn_kv.cpp b/cloud/src/meta-store/mem_txn_kv.cpp index 81755043ea16a0..e4f74337db5e92 100644 --- a/cloud/src/meta-store/mem_txn_kv.cpp +++ b/cloud/src/meta-store/mem_txn_kv.cpp @@ -815,8 +815,6 @@ TxnErrorCode Transaction::batch_get(std::vector>* res auto ret = inner_get(k, &val, opts.snapshot); ret == TxnErrorCode::TXN_OK ? res->push_back(val) : res->push_back(std::nullopt); } - kv_->get_count_ += keys.size(); - num_get_keys_ += keys.size(); return TxnErrorCode::TXN_OK; } diff --git a/cloud/test/meta_service_test.cpp b/cloud/test/meta_service_test.cpp index 4e569bc7ed6a53..4b296e18fc4d3b 100644 --- a/cloud/test/meta_service_test.cpp +++ b/cloud/test/meta_service_test.cpp @@ -20,11 +20,13 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include @@ -2871,6 +2873,11 @@ TEST(MetaServiceTest, AbortTxnWithCoordinatorTest) { TEST(MetaServiceTest, GetPrepareTxnByCoordinatorTest) { auto meta_service = get_meta_service(); + const bool original_mode = config::enable_get_prepare_txn_by_coordinator_by_running_key; + config::enable_get_prepare_txn_by_coordinator_by_running_key = true; + DORIS_CLOUD_DEFER { + config::enable_get_prepare_txn_by_coordinator_by_running_key = original_mode; + }; const int64_t db_id = 888; const int64_t table_id = 999; @@ -2976,6 +2983,447 @@ TEST(MetaServiceTest, GetPrepareTxnByCoordinatorTest) { ASSERT_EQ(resp.status().code(), MetaServiceCode::INVALID_ARGUMENT); } + // Running entries also include precommitted/lazy-committed and expired transactions. + { + std::unique_ptr txn; + ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); + for (int i = 0; i < 2; ++i) { + const auto key = txn_info_key({mock_instance, db_id, txn_ids[i]}); + std::string value; + ASSERT_EQ(txn->get(key, &value), TxnErrorCode::TXN_OK); + TxnInfoPB info; + ASSERT_TRUE(info.ParseFromString(value)); + info.set_status(i == 0 ? TxnStatusPB::TXN_STATUS_PRECOMMITTED + : TxnStatusPB::TXN_STATUS_COMMITTED); + txn->put(key, info.SerializeAsString()); + } + TxnRunningPB expired; + expired.set_timeout_time(0); + txn->put(txn_running_key({mock_instance, db_id, txn_ids[2]}), expired.SerializeAsString()); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + brpc::Controller cntl; + GetPrepareTxnByCoordinatorRequest req; + GetPrepareTxnByCoordinatorResponse resp; + req.set_cloud_unique_id(cloud_unique_id); + req.set_id(coordinator_id); + req.set_ip(host); + req.set_start_time(cur_time + 100); + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); + ASSERT_EQ(resp.txn_infos_size(), 3); + EXPECT_EQ(resp.txn_infos(0).txn_id(), txn_ids[2]); + + // Retry the entire page if its snapshot expires while a candidate is completed/recycled. + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->disable_processing(); + sp->clear_all_call_backs(); + }; + int info_batches = 0; + sp->set_call_back("get_prepare_txn_by_coordinator::batch_get", [&](auto&& args) { + if (++info_batches == 1) { + std::unique_ptr update; + ASSERT_EQ(meta_service->txn_kv()->create_txn(&update), TxnErrorCode::TXN_OK); + update->remove(txn_running_key({mock_instance, db_id, txn_ids[2]})); + update->remove(txn_info_key({mock_instance, db_id, txn_ids[2]})); + ASSERT_EQ(update->commit(), TxnErrorCode::TXN_OK); + *try_any_cast(args[0]) = TxnErrorCode::TXN_TOO_OLD; + } + }); + sp->enable_processing(); + resp.Clear(); + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); + ASSERT_EQ(info_batches, 2); + ASSERT_EQ(resp.txn_infos_size(), 2); + EXPECT_EQ(resp.txn_infos(0).txn_id(), txn_ids[3]); + EXPECT_EQ(resp.txn_infos(1).txn_id(), txn_ids[4]); + + // A running entry without its info in the same snapshot is an error. + ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->remove(txn_info_key({mock_instance, db_id, txn_ids[3]})); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + resp.Clear(); + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + EXPECT_EQ(resp.status().code(), MetaServiceCode::TXN_ID_NOT_FOUND); + } +} + +TEST(MetaServiceTest, GetPrepareTxnByCoordinatorScanRetryTest) { + auto meta_service = get_meta_service(); + const bool original_mode = config::enable_get_prepare_txn_by_coordinator_by_running_key; + const bool original_metrics = config::use_detailed_metrics; + const bool original_retry = config::enable_txn_store_retry; + const auto original_retry_times = config::txn_store_retry_times; + config::txn_store_retry_times = 1; + config::use_detailed_metrics = true; + DORIS_CLOUD_DEFER { + config::enable_get_prepare_txn_by_coordinator_by_running_key = original_mode; + config::use_detailed_metrics = original_metrics; + config::enable_txn_store_retry = original_retry; + config::txn_store_retry_times = original_retry_times; + }; + + GetPrepareTxnByCoordinatorRequest req; + req.set_cloud_unique_id("test_cloud_unique_id"); + req.set_id(12345); + req.set_ip("127.0.0.1"); + for (bool scan_by_running_key : {false, true}) { + config::enable_get_prepare_txn_by_coordinator_by_running_key = scan_by_running_key; + brpc::Controller cntl; + GetPrepareTxnByCoordinatorResponse resp; + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); + EXPECT_EQ(resp.txn_infos_size(), 0); + } + + constexpr int64_t db_id = 888; + constexpr int prepared_count = 5; + std::vector info_bytes; + std::vector running_bytes; + std::unique_ptr txn; + ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); + for (int64_t id = 1; id <= prepared_count + 1; ++id) { + TxnInfoPB info; + info.set_db_id(db_id); + info.set_txn_id(id); + info.set_status(id <= prepared_count ? TxnStatusPB::TXN_STATUS_PREPARED + : TxnStatusPB::TXN_STATUS_VISIBLE); + info.mutable_coordinator()->set_sourcetype(TxnSourceTypePB::TXN_SOURCE_TYPE_BE); + info.mutable_coordinator()->set_id(12345); + info.mutable_coordinator()->set_ip("127.0.0.1"); + const auto key = txn_info_key({mock_instance, db_id, id}); + const auto value = info.SerializeAsString(); + txn->put(key, value); + info_bytes.push_back(key.size() + value.size()); + // Historical info must not be read in running-key mode. + if (id <= prepared_count) { + TxnRunningPB running; + running.set_timeout_time(0); + const auto running_key = txn_running_key({mock_instance, db_id, id}); + const auto running_value = running.SerializeAsString(); + txn->put(running_key, running_value); + running_bytes.push_back(running_key.size() + running_value.size()); + } + } + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + for (const auto& [scan_by_running_key, expire_on_info] : + {std::pair {false, false}, std::pair {true, false}, std::pair {true, true}}) { + const auto& scan_bytes = scan_by_running_key ? running_bytes : info_bytes; + // MemTxnKv needs an empty page after full data pages to finish the scan. + const int terminal_page = scan_bytes.size() + 1; + for (const auto& failures : std::vector> { + {}, {1}, {3}, {terminal_page - 1}, {terminal_page}, {2, 5}}) { + SCOPED_TRACE(fmt::format("scan_by_running_key={} expire_on_info={} failures={}", + scan_by_running_key, expire_on_info, + fmt::join(failures, ","))); + auto [success, message] = + config::set_config({{"enable_get_prepare_txn_by_coordinator_by_running_key", + scan_by_running_key ? "true" : "false"}}, + false, ""); + ASSERT_TRUE(success) << message; + const auto counter_before = + g_bvar_rpc_kv_get_prepare_txn_by_coordinator_get_counter.get({mock_instance}); + const auto bytes_before = + g_bvar_rpc_kv_get_prepare_txn_by_coordinator_get_bytes.get({mock_instance}); + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->disable_processing(); + sp->clear_all_call_backs(); + }; + int pages = 0; + int batches = 0; + sp->set_call_back("memkv::Transaction::get", + [&](auto&& args) { *try_any_cast(args[0]) = 1; }); + sp->set_call_back("get_prepare_txn_by_coordinator::range_get", [&](auto&& args) { + // A config change must only affect subsequent RPCs, even after expiry. + config::enable_get_prepare_txn_by_coordinator_by_running_key = !scan_by_running_key; + ++pages; + if (!expire_on_info && std::count(failures.begin(), failures.end(), pages) != 0) { + *try_any_cast(args[0]) = TxnErrorCode::TXN_TOO_OLD; + } + }); + sp->set_call_back("get_prepare_txn_by_coordinator::batch_get", [&](auto&& args) { + ++batches; + if (expire_on_info && std::count(failures.begin(), failures.end(), batches) != 0) { + *try_any_cast(args[0]) = TxnErrorCode::TXN_TOO_OLD; + } + }); + sp->enable_processing(); + + brpc::Controller cntl; + GetPrepareTxnByCoordinatorResponse resp; + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); + ASSERT_EQ(resp.txn_infos_size(), prepared_count); + for (int i = 0; i < prepared_count; ++i) { + EXPECT_EQ(resp.txn_infos(i).txn_id(), i + 1); + } + EXPECT_EQ(pages, terminal_page + failures.size()); + EXPECT_EQ(batches, scan_by_running_key + ? terminal_page + (expire_on_info ? failures.size() : 0) + : 0); + + // Each earlier retry shifts subsequent read-attempt numbers by one. + std::vector failed_pages; + for (size_t i = 0; i < failures.size(); ++i) { + failed_pages.push_back(failures[i] - static_cast(i)); + } + int64_t expected_counter = 0; + int64_t expected_bytes = 0; + for (size_t i = 0; i < scan_bytes.size(); ++i) { + const int reads = 1 + std::count(failed_pages.begin(), failed_pages.end(), i + 1); + expected_counter += reads; + expected_bytes += reads * scan_bytes[i]; + } + if (scan_by_running_key) { + for (int i = 0; i < prepared_count; ++i) { + const int reads = 1 + (expire_on_info ? std::count(failed_pages.begin(), + failed_pages.end(), i + 1) + : 0); + expected_counter += reads; + expected_bytes += reads * info_bytes[i]; + } + } + EXPECT_EQ( + g_bvar_rpc_kv_get_prepare_txn_by_coordinator_get_counter.get({mock_instance}) - + counter_before, + expected_counter); + EXPECT_EQ(g_bvar_rpc_kv_get_prepare_txn_by_coordinator_get_bytes.get({mock_instance}) - + bytes_before, + expected_bytes); + } + + // Fail after two pages have already appended results. Check handler failure separately + // from the proxy retry, which must clear partial results before restarting the RPC. + for (const auto& [error, retry_whole_rpc] : + {std::pair {TxnErrorCode::TXN_TOO_OLD, false}, + std::pair {TxnErrorCode::TXN_UNIDENTIFIED_ERROR, false}, + std::pair {TxnErrorCode::TXN_TOO_OLD, true}}) { + SCOPED_TRACE( + fmt::format("scan_by_running_key={} expire_on_info={} error={} proxy_retry={}", + scan_by_running_key, expire_on_info, error, retry_whole_rpc)); + config::enable_get_prepare_txn_by_coordinator_by_running_key = scan_by_running_key; + config::enable_txn_store_retry = retry_whole_rpc; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->disable_processing(); + sp->clear_all_call_backs(); + }; + int attempts = 0; + sp->set_call_back("memkv::Transaction::get", + [&](auto&& args) { *try_any_cast(args[0]) = 1; }); + sp->set_call_back(expire_on_info ? "get_prepare_txn_by_coordinator::batch_get" + : "get_prepare_txn_by_coordinator::range_get", + [&](auto&& args) { + ++attempts; + if (attempts == 3 || attempts == 4) { + *try_any_cast(args[0]) = error; + } + }); + sp->enable_processing(); + brpc::Controller cntl; + GetPrepareTxnByCoordinatorResponse resp; + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + if (retry_whole_rpc) { + ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); + EXPECT_EQ(attempts, 4 + terminal_page); + ASSERT_EQ(resp.txn_infos_size(), prepared_count); + for (int i = 0; i < prepared_count; ++i) { + EXPECT_EQ(resp.txn_infos(i).txn_id(), i + 1); + } + } else { + EXPECT_EQ(resp.status().code(), error == TxnErrorCode::TXN_TOO_OLD + ? MetaServiceCode::KV_TXN_TOO_OLD + : MetaServiceCode::KV_TXN_GET_ERR); + EXPECT_EQ(attempts, error == TxnErrorCode::TXN_TOO_OLD ? 4 : 3); + EXPECT_EQ(resp.status().msg(), + fmt::format( + "get_prepare_txn_by_coordinator: failed to get txn info. err={}", + error)); + } + } + } +} + +TEST(MetaServiceTest, GetPrepareTxnByCoordinatorScanModeTest) { + auto meta_service = get_meta_service(); + const bool original_mode = config::enable_get_prepare_txn_by_coordinator_by_running_key; + DORIS_CLOUD_DEFER { + config::enable_get_prepare_txn_by_coordinator_by_running_key = original_mode; + }; + + constexpr int64_t db_id = 888; + std::unique_ptr txn; + ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); + for (int64_t id = 1; id <= 10; ++id) { + TxnInfoPB info; + info.set_db_id(db_id); + info.set_txn_id(id); + info.set_status(id == 6 ? TxnStatusPB::TXN_STATUS_VISIBLE + : id == 7 ? TxnStatusPB::TXN_STATUS_PRECOMMITTED + : id == 8 ? TxnStatusPB::TXN_STATUS_COMMITTED + : TxnStatusPB::TXN_STATUS_PREPARED); + auto* coordinator = info.mutable_coordinator(); + coordinator->set_sourcetype(id == 4 ? TxnSourceTypePB::TXN_SOURCE_TYPE_FE + : TxnSourceTypePB::TXN_SOURCE_TYPE_BE); + coordinator->set_id(id == 2 ? 0 : id == 9 ? 54321 : 12345); + coordinator->set_ip(id == 3 ? "127.0.0.2" : "127.0.0.1"); + coordinator->set_start_time(id == 5 ? 200 : id == 10 ? 150 : 100); + const auto key = txn_info_key({mock_instance, db_id, id}); + const auto value = info.SerializeAsString(); + txn->put(key, value); + if (id != 6) { + TxnRunningPB running; + running.set_timeout_time(0); + const auto running_key = txn_running_key({mock_instance, db_id, id}); + const auto running_value = running.SerializeAsString(); + txn->put(running_key, running_value); + } + } + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + for (bool scan_by_running_key : {false, true}) { + SCOPED_TRACE(fmt::format("scan_by_running_key={}", scan_by_running_key)); + config::enable_get_prepare_txn_by_coordinator_by_running_key = scan_by_running_key; + brpc::Controller cntl; + GetPrepareTxnByCoordinatorRequest req; + GetPrepareTxnByCoordinatorResponse resp; + req.set_cloud_unique_id("test_cloud_unique_id"); + req.set_id(12345); + req.set_ip("127.0.0.1"); + req.set_start_time(150); + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); + ASSERT_EQ(resp.txn_infos_size(), 2); + EXPECT_EQ(resp.txn_infos(0).txn_id(), 1); + EXPECT_EQ(resp.txn_infos(1).txn_id(), 2); + } + + // Skip malformed running keys on the first, middle and last data pages. + { + ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); + for (int64_t id : {0, 3, 11}) { + auto key = txn_running_key({mock_instance, db_id, id}); + key.push_back('\xff'); + txn->put(key, ""); + } + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + config::enable_get_prepare_txn_by_coordinator_by_running_key = true; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->disable_processing(); + sp->clear_all_call_backs(); + }; + int pages = 0; + sp->set_call_back("memkv::Transaction::get", [&](auto&& args) { + *try_any_cast(args[0]) = 1; + ++pages; + }); + sp->enable_processing(); + + brpc::Controller cntl; + GetPrepareTxnByCoordinatorRequest req; + GetPrepareTxnByCoordinatorResponse resp; + req.set_cloud_unique_id("test_cloud_unique_id"); + req.set_id(12345); + req.set_ip("127.0.0.1"); + req.set_start_time(150); + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); + ASSERT_EQ(resp.txn_infos_size(), 2); + EXPECT_EQ(resp.txn_infos(0).txn_id(), 1); + EXPECT_EQ(resp.txn_infos(1).txn_id(), 2); + EXPECT_EQ(pages, 9 + 3 + 1); + } + + // Processing inside read_page must preserve the RPC parse error in both scan modes. + ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); + const auto malformed_key = txn_info_key({mock_instance, db_id, 1}); + txn->put(malformed_key, std::string(1, '\xff')); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + for (bool scan_by_running_key : {false, true}) { + config::enable_get_prepare_txn_by_coordinator_by_running_key = scan_by_running_key; + brpc::Controller cntl; + GetPrepareTxnByCoordinatorRequest req; + GetPrepareTxnByCoordinatorResponse resp; + req.set_cloud_unique_id("test_cloud_unique_id"); + req.set_id(12345); + req.set_ip("127.0.0.1"); + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + EXPECT_EQ(resp.status().code(), MetaServiceCode::PROTOBUF_PARSE_ERR); + EXPECT_EQ(resp.status().msg(), "malformed txn info, key=" + hex(malformed_key)); + } +} + +TEST(MetaServiceTest, GetPrepareTxnByCoordinatorLargeInfoTest) { + auto meta_service = get_meta_service(); + const bool original_mode = config::enable_get_prepare_txn_by_coordinator_by_running_key; + config::enable_get_prepare_txn_by_coordinator_by_running_key = true; + DORIS_CLOUD_DEFER { + config::enable_get_prepare_txn_by_coordinator_by_running_key = original_mode; + }; + constexpr int txn_count = 257; + constexpr int64_t db_id = 888; + const std::string large_reason(50 * 1024, 'x'); + for (int64_t id = 1; id <= txn_count; ++id) { + TxnInfoPB info; + info.set_db_id(db_id); + info.set_txn_id(id); + info.set_status(TxnStatusPB::TXN_STATUS_PREPARED); + info.set_reason(large_reason); + info.mutable_coordinator()->set_sourcetype(TxnSourceTypePB::TXN_SOURCE_TYPE_BE); + info.mutable_coordinator()->set_id(12345); + info.mutable_coordinator()->set_ip("127.0.0.1"); + TxnRunningPB running; + running.set_timeout_time(0); + + std::unique_ptr txn; + ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->put(txn_info_key({mock_instance, db_id, id}), info.SerializeAsString()); + txn->put(txn_running_key({mock_instance, db_id, id}), running.SerializeAsString()); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + } + + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->disable_processing(); + sp->clear_all_call_backs(); + }; + int pages = 0; + size_t read_count = 0; + sp->set_call_back("get_prepare_txn_by_coordinator::batch_get", [&](auto&& args) { + const auto& values = *try_any_cast>*>(args[1]); + EXPECT_LE(values.size(), 128); + size_t bytes = 0; + for (const auto& value : values) { + ASSERT_TRUE(value.has_value()); + EXPECT_GE(value->size(), large_reason.size()); + bytes += value->size(); + } + EXPECT_LT(bytes, 7 * 1024 * 1024); + read_count += values.size(); + ++pages; + }); + sp->enable_processing(); + + brpc::Controller cntl; + GetPrepareTxnByCoordinatorRequest req; + GetPrepareTxnByCoordinatorResponse resp; + req.set_cloud_unique_id("test_cloud_unique_id"); + req.set_id(12345); + req.set_ip("127.0.0.1"); + meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp, nullptr); + ASSERT_EQ(resp.status().code(), MetaServiceCode::OK); + EXPECT_EQ(pages, 3); + EXPECT_EQ(read_count, txn_count); + ASSERT_EQ(resp.txn_infos_size(), txn_count); + for (int i = 0; i < txn_count; ++i) { + EXPECT_EQ(resp.txn_infos(i).txn_id(), i + 1); + EXPECT_EQ(resp.txn_infos(i).reason(), large_reason); + } } TEST(MetaServiceTest, CheckTxnConflictTest) { From 52dbed63a04aa52fb3276f5fd45d83df98dc69cc Mon Sep 17 00:00:00 2001 From: meiyi Date: Thu, 10 Sep 2026 16:38:06 +0800 Subject: [PATCH 2/2] [improvement](cloud) Log coordinator scan mode and reserve keys ### What problem does this PR solve? Problem Summary: Include the scan mode in the initial coordinator cleanup log so failed scans also identify their strategy. Reserve txn-info key capacity using the current running-key page size to avoid vector growth. ### Release note None ### Check List (For Author) - Test: Formatting and git diff --check passed; no build or tests run as requested. - Behavior changed: No; logging and allocation only. - Does this need documentation: No --- cloud/src/meta-service/meta_service_txn.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index 135a2c38837f2a..de3cfb26c78237 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/cloud/src/meta-service/meta_service_txn.cpp @@ -4665,7 +4665,8 @@ void MetaServiceImpl::get_prepare_txn_by_coordinator( : txn_info_key({instance_id, 0, 0}); std::string end_key = scan_by_running_key ? txn_running_key({instance_id, INT64_MAX, INT64_MAX}) : txn_info_key({instance_id, INT64_MAX, INT64_MAX}); - LOG(INFO) << "begin_key:" << hex(begin_key) << " end_key:" << hex(end_key); + LOG(INFO) << "begin_key:" << hex(begin_key) << " end_key:" << hex(end_key) + << " scan_by_running_key=" << scan_by_running_key; TxnErrorCode err = txn_kv_->create_txn(&txn); if (err != TxnErrorCode::TXN_OK) { @@ -4714,6 +4715,7 @@ void MetaServiceImpl::get_prepare_txn_by_coordinator( return ret; } std::vector info_keys; + info_keys.reserve(scan_by_running_key ? it->size() : 0); while (it->has_next()) { auto [key, value] = it->next(); if (scan_by_running_key) {