Skip to content

[fix](cloud) Only mark prepare rowsets before recycle #65550

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

Open
wyxxxcat wants to merge 1 commit into
base: master
Choose a base branch
from
+46 −110
Open

[fix](cloud) Only mark prepare rowsets before recycle #65550

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
153 changes: 46 additions & 107 deletions cloud/src/recycler/recycler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1954,15 +1954,9 @@ struct DeferredRecyclePrepareDeleteTask {
int64_t tablet_id = 0;
};

template <typename T>
std::optional<DeferredRecycleAbortTask> make_deferred_abort_task(const T& rowset_meta_pb) {
if constexpr (std::is_same_v<T, RecycleRowsetPB>) {
if (rowset_meta_pb.type() != RecycleRowsetPB::PREPARE) {
return std::nullopt;
}
}

const auto& rs_meta = rowset_meta(rowset_meta_pb);
std::optional<DeferredRecycleAbortTask> make_deferred_abort_task(
const RowsetMetaCloudPB& rowset_meta_pb) {
const auto& rs_meta = rowset_meta_pb;
DeferredRecycleAbortTask task;
task.tablet_id = rs_meta.tablet_id();
task.start_version = rs_meta.start_version();
Expand All @@ -1981,10 +1975,8 @@ std::optional<DeferredRecycleAbortTask> make_deferred_abort_task(const T& rowset
return std::nullopt;
}

template <typename T>
bool need_mark_rowset_as_recycled(const T& rowset_meta_pb) {
const auto& rs_meta = rowset_meta(rowset_meta_pb);
return !rs_meta.has_is_recycled() || !rs_meta.is_recycled();
bool need_mark_rowset_as_recycled(const RowsetMetaCloudPB& rowset_meta_pb) {
return !rowset_meta_pb.has_is_recycled() || !rowset_meta_pb.is_recycled();
}

template <typename T>
Expand Down Expand Up @@ -2017,7 +2009,7 @@ int batch_mark_rowsets_as_recycled(TxnKv* txn_kv, const std::string& instance_id
<< " key=" << hex(key);
return -1;
}
if (!need_mark_rowset_as_recycled(rowset_meta_pb)) {
if (!need_mark_rowset_as_recycled(rowset_meta(rowset_meta_pb))) {
continue;
}
mutable_rowset_meta(rowset_meta_pb)->set_is_recycled(true);
Expand Down Expand Up @@ -2070,7 +2062,12 @@ int collect_deferred_abort_tasks(TxnKv* txn_kv, const std::string& instance_id,
if (skip_base_version && rowset_meta(rowset_meta_pb).end_version() == 1) {
continue;
}
if (auto abort_task = make_deferred_abort_task(rowset_meta_pb);
if constexpr (std::is_same_v<T, RecycleRowsetPB>) {
if (rowset_meta_pb.type() != RecycleRowsetPB::PREPARE) {
continue;
}
}
if (auto abort_task = make_deferred_abort_task(rowset_meta(rowset_meta_pb));
abort_task.has_value()) {
abort_tasks->emplace_back(std::move(*abort_task));
}
Expand Down Expand Up @@ -2112,48 +2109,6 @@ int InstanceRecycler::batch_abort_txn_or_job_for_recycle(const std::vector<std::
return 0;
}

int collect_prepare_delete_tasks(TxnKv* txn_kv, const std::string& instance_id,
const std::vector<std::string>& keys,
std::vector<DeferredRecyclePrepareDeleteTask>* delete_tasks) {
constexpr size_t kPrepareCheckBatchSize = 256;
for (size_t offset = 0; offset < keys.size(); offset += kPrepareCheckBatchSize) {
size_t limit = std::min(keys.size(), offset + kPrepareCheckBatchSize);
std::unique_ptr<Transaction> txn;
TxnErrorCode err = txn_kv->create_txn(&txn);
if (err != TxnErrorCode::TXN_OK) {
LOG(WARNING) << "failed to create txn, instance_id=" << instance_id;
return -1;
}
for (size_t idx = offset; idx < limit; ++idx) {
const std::string& key = keys[idx];
std::string val;
err = txn->get(key, &val);
if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
// has already been removed
continue;
}
if (err != TxnErrorCode::TXN_OK) {
LOG(WARNING) << "failed to get recycle rowset, instance_id=" << instance_id
<< " key=" << hex(key);
return -1;
}
RecycleRowsetPB rowset;
if (!rowset.ParseFromString(val)) {
LOG(WARNING) << "failed to parse recycle rowset, instance_id=" << instance_id
<< " key=" << hex(key);
return -1;
}
if (rowset.type() != RecycleRowsetPB::PREPARE) {
continue;
}
const auto& rs_meta = rowset.rowset_meta();
delete_tasks->push_back(
{key, rs_meta.resource_id(), rs_meta.rowset_id_v2(), rs_meta.tablet_id()});
}
}
return 0;
}

int InstanceRecycler::recycle_ref_rowsets(bool* has_unrecycled_rowsets) {
const std::string task_name = "recycle_ref_rowsets";
*has_unrecycled_rowsets = false;
Expand Down Expand Up @@ -5000,7 +4955,6 @@ int InstanceRecycler::recycle_rowsets() {
};

std::vector<std::string> rowset_keys_to_mark_recycled;
std::vector<std::string> rowset_keys_to_abort;
std::vector<std::string> prepare_rowset_keys_to_delete;

// Store keys of rowset recycled by background workers
Expand All @@ -5010,9 +4964,6 @@ int InstanceRecycler::recycle_rowsets() {
auto worker_pool = std::make_unique<SimpleThreadPool>(
config::instance_recycler_worker_pool_size, "recycle_rowsets");
worker_pool->start();
auto mark_abort_worker_pool = std::make_unique<SimpleThreadPool>(
config::instance_recycler_worker_pool_size, "recycle_rs_mark_abort");
mark_abort_worker_pool->start();
// TODO bacth delete
auto delete_versioned_delete_bitmap_kvs = [&](int64_t tablet_id, const std::string& rowset_id) {
std::string dbm_start_key =
Expand Down Expand Up @@ -5132,31 +5083,7 @@ int InstanceRecycler::recycle_rowsets() {
segment_metrics_context_.total_recycled_num += rowset.rowset_meta().num_segments();
return 0;
}

auto* rowset_meta = rowset.mutable_rowset_meta();
if (config::enable_mark_delete_rowset_before_recycle) {
if (need_mark_rowset_as_recycled(rowset)) {
rowset_keys_to_mark_recycled.emplace_back(k);
LOG(INFO) << "rowset queued to mark as recycled, recycler will delete data and kv "
"at next turn, instance_id="
<< instance_id_ << " tablet_id=" << rowset_meta->tablet_id()
<< " version=[" << rowset_meta->start_version() << '-'
<< rowset_meta->end_version() << "]";
return 0;
}
}

if (config::enable_abort_txn_and_job_for_delete_rowset_before_recycle &&
rowset_meta->end_version() != 1) {
if (make_deferred_abort_task(rowset).has_value()) {
LOG(INFO) << "rowset queued to abort related txn or job after current scan batch, "
"instance_id="
<< instance_id_ << " tablet_id=" << rowset_meta->tablet_id()
<< " version=[" << rowset_meta->start_version() << '-'
<< rowset_meta->end_version() << "]";
rowset_keys_to_abort.emplace_back(k);
}
}

// TODO(plat1ko): check rowset not referenced
if (!rowset_meta->has_resource_id()) [[unlikely]] { // impossible
Expand All @@ -5178,6 +5105,35 @@ int InstanceRecycler::recycle_rowsets() {
<< " creation_time=" << rowset_meta->creation_time()
<< " task_type=" << metrics_context.operation_type;
if (rowset.type() == RecycleRowsetPB::PREPARE) {
if (config::enable_mark_delete_rowset_before_recycle) {
if (need_mark_rowset_as_recycled(rowset.rowset_meta())) {
rowset_keys_to_mark_recycled.emplace_back(k);
LOG(INFO) << "rowset queued to mark as recycled, recycler will delete data and "
"kv "
"at next turn, instance_id="
<< instance_id_ << " tablet_id=" << rowset_meta->tablet_id()
<< " version=[" << rowset_meta->start_version() << '-'
<< rowset_meta->end_version() << "]";
return 0;
}
}

if (config::enable_abort_txn_and_job_for_delete_rowset_before_recycle &&
rowset_meta->end_version() != 1) {
int ret = 0;
if (rowset_meta->has_load_id()) {
DCHECK(rowset_meta->has_txn_id() && rowset_meta->txn_id() > 0);
ret = abort_txn_for_related_rowset(rowset_meta->txn_id());
} else if (rowset_meta->has_job_id()) {
ret = abort_job_for_related_rowset(*rowset_meta);
}
if (ret != 0) {
LOG(WARNING) << "failed to abort txn or job for related rowset, instance_id="
<< instance_id_ << " tablet_id=" << rowset_meta->tablet_id()
<< " version=[" << rowset_meta->start_version() << '-'
<< rowset_meta->end_version() << "]";
}
}
// unable to calculate file path, can only be deleted by rowset id prefix
num_prepare += 1;
if (delete_rowset_data_by_prefix(std::string(k), rowset_meta->resource_id(),
Expand Down Expand Up @@ -5221,21 +5177,17 @@ int InstanceRecycler::recycle_rowsets() {
});
};

auto submit_mark_abort_rowset_job = [&](std::vector<std::string> rowset_keys_to_mark,
std::vector<std::string> rowset_keys_to_abort) {
if (rowset_keys_to_mark.empty() && rowset_keys_to_abort.empty()) {
auto submit_mark_rowset_job = [&](std::vector<std::string> rowset_keys_to_mark) {
if (rowset_keys_to_mark.empty()) {
return;
}
mark_abort_worker_pool->submit([&, rowset_keys_to_mark = std::move(rowset_keys_to_mark),
rowset_keys_to_abort =
std::move(rowset_keys_to_abort)]() mutable {
worker_pool->submit([&, rowset_keys_to_mark = std::move(rowset_keys_to_mark)]() mutable {
auto start = steady_clock::now();
DORIS_CLOUD_DEFER {
auto cost = duration_cast<milliseconds>(steady_clock::now() - start).count();
LOG(INFO) << "finish mark and abort rowset job, instance_id=" << instance_id_ << ' '
<< "cost_ms=" << cost << ' '
<< "rowset_keys_to_mark.size()=" << rowset_keys_to_mark.size() << ' '
<< "rowset_keys_to_abort.size()=" << rowset_keys_to_abort.size();
<< "rowset_keys_to_mark.size()=" << rowset_keys_to_mark.size();
};
if (!rowset_keys_to_mark.empty() &&
batch_mark_rowsets_as_recycled<RecycleRowsetPB>(txn_kv_.get(), instance_id_,
Expand All @@ -5245,26 +5197,14 @@ int InstanceRecycler::recycle_rowsets() {
<< "rowset_keys_to_mark.size()=" << rowset_keys_to_mark.size();
return;
}
if (!rowset_keys_to_abort.empty() &&
batch_abort_txn_or_job_for_recycle<RecycleRowsetPB>(rowset_keys_to_abort, true) !=
0) {
LOG(WARNING) << "failed to batch abort txn or job for related rowset, "
"instance_id="
<< instance_id_ << ' '
<< "rowset_keys_to_abort.size()=" << rowset_keys_to_abort.size();
return;
}
});
};

bool scan_finished = false;
auto loop_done = [&]() -> int {
std::vector<std::string> mark_keys_to_process;
std::vector<std::string> abort_keys_to_process;
mark_keys_to_process.swap(rowset_keys_to_mark_recycled);
abort_keys_to_process.swap(rowset_keys_to_abort);
submit_mark_abort_rowset_job(std::move(mark_keys_to_process),
std::move(abort_keys_to_process));
submit_mark_rowset_job(std::move(mark_keys_to_process));
if (!scan_finished && rowsets.size() < delete_rowset_batch_size) {
return 0;
}
Expand Down Expand Up @@ -5332,7 +5272,6 @@ int InstanceRecycler::recycle_rowsets() {
ret = -1;
}

mark_abort_worker_pool->stop();
worker_pool->stop();

if (!async_recycled_rowset_keys.empty()) {
Expand Down
3 changes: 0 additions & 3 deletions cloud/test/recycler_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1648,17 +1648,14 @@ TEST(RecyclerTest, recycle_rowsets_limit_per_tablet_batch) {
return it->size();
};

ASSERT_EQ(recycler.recycle_rowsets(), 0);
ASSERT_EQ(recycler.recycle_rowsets(), 0);
EXPECT_EQ(count_recycle_rowsets(tablet_id0), 3);
EXPECT_EQ(count_recycle_rowsets(tablet_id1), 3);

ASSERT_EQ(recycler.recycle_rowsets(), 0);
ASSERT_EQ(recycler.recycle_rowsets(), 0);
EXPECT_EQ(count_recycle_rowsets(tablet_id0), 1);
EXPECT_EQ(count_recycle_rowsets(tablet_id1), 1);

ASSERT_EQ(recycler.recycle_rowsets(), 0);
ASSERT_EQ(recycler.recycle_rowsets(), 0);
EXPECT_EQ(count_recycle_rowsets(tablet_id0), 0);
EXPECT_EQ(count_recycle_rowsets(tablet_id1), 0);
Expand Down
Loading