Skip to content

branch-4.1: [fix](cloud) Avoid conflicts when recycling transactions with the same label #65746 #67478

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiguolei merged 1 commit into from
Sep 5, 2026
+280 −55
Merged

branch-4.1: [fix](cloud) Avoid conflicts when recycling transactions with the same label #65746 #67478

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 65 additions & 39 deletions cloud/src/recycler/recycler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6706,7 +6706,7 @@ int InstanceRecycler::recycle_expired_txn_label() {
std::string end_recycle_txn_key;
recycle_txn_key(recycle_txn_key_info0, &begin_recycle_txn_key);
recycle_txn_key(recycle_txn_key_info1, &end_recycle_txn_key);
std::vector<std::string> recycle_txn_info_keys;
std::unordered_map<std::string, std::vector<std::string>> recycle_txn_keys_by_label;

LOG_WARNING("begin to recycle expired txn").tag("instance_id", instance_id_);

Expand Down Expand Up @@ -6747,7 +6747,17 @@ int InstanceRecycler::recycle_expired_txn_label() {
current_time_ms)) {
VLOG_DEBUG << "found recycle txn, key=" << hex(k);
num_expired++;
recycle_txn_info_keys.emplace_back(k);

std::string_view k1 = k;
k1.remove_prefix(1); // Remove key space
std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> out;
if (decode_key(&k1, &out) != 0) {
LOG_ERROR("failed to decode key").tag("key", hex(k));
return -1;
}
int64_t db_id = std::get<int64_t>(std::get<0>(out[3]));
auto label_key = txn_label_key({instance_id_, db_id, recycle_txn_pb.label()});
recycle_txn_keys_by_label[label_key].emplace_back(k);
}
return 0;
};
Expand Down Expand Up @@ -6856,35 +6866,48 @@ int InstanceRecycler::recycle_expired_txn_label() {

auto loop_done = [&]() -> int {
DORIS_CLOUD_DEFER {
recycle_txn_info_keys.clear();
recycle_txn_keys_by_label.clear();
};
TEST_SYNC_POINT_CALLBACK(
"InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_info_keys",
&recycle_txn_info_keys);
for (const auto& k : recycle_txn_info_keys) {
concurrent_delete_executor.add([&]() {
int ret = delete_recycle_txn_kv(k);
if (ret == 1) {
const int max_retry = std::max(1, config::recycle_txn_delete_max_retry_times);
for (int i = 1; i <= max_retry; ++i) {
LOG(WARNING) << "txn conflict, retry times=" << i << " key=" << hex(k);
ret = delete_recycle_txn_kv(k);
// clang-format off
TEST_SYNC_POINT_CALLBACK(
"InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error", &ret);
// clang-format off
if (ret != 1) {
break;
}
// random sleep 0-100 ms to retry
std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100));
"InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_keys_by_label",
&recycle_txn_keys_by_label);
auto delete_recycle_txn_kv_with_retry = [&](const std::string& k) -> int {
int ret = delete_recycle_txn_kv(k);
TEST_SYNC_POINT_CALLBACK(
"InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error",
&ret);
if (ret == 1) {
const int max_retry = std::max(1, config::recycle_txn_delete_max_retry_times);
for (int i = 1; i <= max_retry; ++i) {
LOG(WARNING) << "txn conflict, retry times=" << i << " key=" << hex(k);
ret = delete_recycle_txn_kv(k);
TEST_SYNC_POINT_CALLBACK(
"InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_"
"error",
&ret);
if (ret != 1) {
break;
}
// random sleep 0-100 ms to retry
std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100));
}
if (ret != 0) {
LOG_WARNING("failed to delete recycle txn kv")
.tag("instance id", instance_id_)
.tag("key", hex(k));
return -1;
}
return ret;
};

for (auto& [label_key, txn_keys] : recycle_txn_keys_by_label) {
concurrent_delete_executor.add([&, txn_keys = std::move(txn_keys), label_key]() {
VLOG_DEBUG << "recycle txn label group, key=" << hex(label_key)
<< " txn_count=" << txn_keys.size();
for (const auto& k : txn_keys) {
int ret = delete_recycle_txn_kv_with_retry(k);
if (ret != 0) {
LOG_WARNING("failed to delete recycle txn kv")
.tag("instance id", instance_id_)
.tag("key", hex(k))
.tag("label_key", hex(label_key));
return -1;
}
}
return 0;
});
Expand Down Expand Up @@ -7646,8 +7669,8 @@ int InstanceRecycler::scan_and_statistics_rowsets() {
std::string recyc_rs_key0;
std::string recyc_rs_key1;
recycle_rowset_key(recyc_rs_key_info0, &recyc_rs_key0);
recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1);
int64_t earlest_ts = std::numeric_limits<int64_t>::max();
recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1);
int64_t earlest_ts = std::numeric_limits<int64_t>::max();

auto handle_rowset_kv = [&, this](std::string_view k, std::string_view v) -> int {
RecycleRowsetPB rowset;
Expand All @@ -7671,7 +7694,8 @@ int InstanceRecycler::scan_and_statistics_rowsets() {
metrics_context.total_need_recycle_num++;
metrics_context.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size();
segment_metrics_context_.total_need_recycle_num += rowset.rowset_meta().num_segments();
segment_metrics_context_.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size();
segment_metrics_context_.total_need_recycle_data_size +=
rowset.rowset_meta().total_disk_size();
return 0;
}

Expand Down Expand Up @@ -7724,7 +7748,7 @@ int InstanceRecycler::scan_and_statistics_tmp_rowsets() {
DCHECK_GT(rowset.txn_id(), 0)
<< "txn_id=" << rowset.txn_id() << " rowset=" << rowset.ShortDebugString();

if(!rowset.has_is_recycled() || !rowset.is_recycled()) {
if (!rowset.has_is_recycled() || !rowset.is_recycled()) {
return 0;
}

Expand Down Expand Up @@ -7801,7 +7825,8 @@ int InstanceRecycler::scan_and_statistics_abort_timeout_txn() {
return 0;
};

int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key, std::move(handle_abort_timeout_txn_kv));
int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key,
std::move(handle_abort_timeout_txn_kv));
metrics_context.report(true);
return ret;
}
Expand Down Expand Up @@ -7835,7 +7860,8 @@ int InstanceRecycler::scan_and_statistics_expired_txn_label() {
return 0;
};

int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key, std::move(handle_expired_txn_label_kv));
int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key,
std::move(handle_expired_txn_label_kv));
metrics_context.report(true);
return ret;
}
Expand Down Expand Up @@ -8066,7 +8092,7 @@ int InstanceRecycler::scan_and_statistics_restore_jobs() {
return 0;
}
metrics_context.total_need_recycle_num++;
if(restore_job_pb.need_recycle_data()) {
if (restore_job_pb.need_recycle_data()) {
scan_tablet_and_statistics(restore_job_pb.tablet_id(), metrics_context);
}
return 0;
Expand Down Expand Up @@ -8108,7 +8134,7 @@ void InstanceRecycler::scan_and_statistics_operation_logs() {

OperationLogReferenceInfo ref_info;
if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(),
&ref_info)) {
&ref_info)) {
metrics_context.total_need_recycle_num++;
metrics_context.total_need_recycle_data_size += operation_log.ByteSizeLong();
}
Expand Down Expand Up @@ -8264,8 +8290,8 @@ int InstanceRecycler::cleanup_rowset_metadata(const std::vector<RowsetDeleteTask
std::string dbm_start_key =
meta_delete_bitmap_key({reference_instance_id, tablet_id, rowset_id, 0, 0});
std::string dbm_end_key = meta_delete_bitmap_key(
{reference_instance_id, tablet_id, rowset_id,
std::numeric_limits<int64_t>::max(), std::numeric_limits<int64_t>::max()});
{reference_instance_id, tablet_id, rowset_id, std::numeric_limits<int64_t>::max(),
std::numeric_limits<int64_t>::max()});
txn->remove(dbm_start_key, dbm_end_key);
LOG_INFO("remove delete bitmap kv in cleanup phase")
.tag("instance_id", instance_id_)
Expand All @@ -8288,8 +8314,8 @@ int InstanceRecycler::cleanup_rowset_metadata(const std::vector<RowsetDeleteTask

// Remove versioned meta rowset key
if (!task.versioned_rowset_key.empty()) {
versioned::document_remove<RowsetMetaCloudPB>(
txn.get(), task.versioned_rowset_key, task.versionstamp);
versioned::document_remove<RowsetMetaCloudPB>(txn.get(), task.versioned_rowset_key,
task.versionstamp);
LOG_INFO("remove versioned meta rowset key in cleanup phase")
.tag("instance_id", instance_id_)
.tag("tablet_id", tablet_id)
Expand Down
Loading
Loading