github-actions[bot] commented on code in PR #65550:
URL: https://github.com/apache/doris/pull/65550#discussion_r3780954829
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -5067,6 +5189,33 @@ int InstanceRecycler::recycle_rowsets() {
return 0;
};
+ auto submit_recycle_prepare_rowsets_job = [&](std::string key, std::string
resource_id,
+ int64_t tablet_id,
std::string rowset_id) {
+ int ret = worker_pool->submit([&, key = std::move(key),
+ resource_id = std::move(resource_id),
tablet_id,
+ rowset_id = std::move(rowset_id)]()
mutable {
+ std::vector<std::string> aborted_keys;
+ if (batch_abort_txn_or_job_for_recycle<RecycleRowsetPB>({key},
aborted_keys) != 0 ||
+ aborted_keys.empty()) {
+ return;
+ }
+ if (delete_rowset_data(resource_id, tablet_id, rowset_id) != 0) {
+ LOG(WARNING) << "failed to delete rowset data, key=" <<
hex(key);
+ return;
+ }
+ if (delete_versioned_delete_bitmap_kvs(tablet_id, rowset_id) != 0)
{
+ return;
+ }
+ if (txn_remove(txn_kv_.get(), aborted_keys) != 0) {
Review Comment:
[P2] Count successful abort-protected deletions
After this removal succeeds, the new PREPARE path exits without incrementing
`num_recycled` as the other prefix/formal deletion workers do. An all-PREPARE
abort-enabled run can therefore delete every object and KV while its completion
log reports `num_recycled=0`. Update the counter after successful KV removal
and cover it in the new PREPARE tests.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -5177,8 +5302,34 @@ int InstanceRecycler::recycle_rowsets() {
<< " creation_time=" << rowset_meta->creation_time()
<< " task_type=" << metrics_context.operation_type;
if (rowset.type() == RecycleRowsetPB::PREPARE) {
- // unable to calculate file path, can only be deleted by rowset id
prefix
+ 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;
+ }
+ }
+
num_prepare += 1;
+ if
(config::enable_abort_txn_and_job_for_delete_rowset_before_recycle &&
+ rowset_meta->end_version() != 1) {
+ if (make_related_txn_or_job_abort_task(rowset).has_value()) {
+ LOG(INFO) << "rowset queued to abort related txn or job
before recycling, "
+ "instance_id="
+ << instance_id_ << " tablet_id=" <<
rowset_meta->tablet_id()
+ << " version=[" << rowset_meta->start_version()
<< '-'
+ << rowset_meta->end_version() << "]";
+ submit_recycle_prepare_rowsets_job(std::string(k),
rowset_meta->resource_id(),
Review Comment:
[P2] Coalesce PREPARE aborts by owner
This submits one abort transaction per rowset, so the default 32 workers
race when many PREPARE rowsets share one load txn or tablet job. They all
read/write the same owner record; one commit wins and the losers return
`TXN_CONFLICT`, retaining already-expired keys until the next recycler interval
(normally 3600s). Group by `txn_id` or exact `(tablet_id, job_id)`, abort once,
and fan the result out to all owned keys. The added 257-key test does not cover
this: an earlier test leaks `worker_pool_size=1`, and each invocation here
passes only `{key}`, so it neither exercises production concurrency nor the
256-item collector batch.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -5990,12 +6104,13 @@ int InstanceRecycler::recycle_tmp_rowsets() {
}
if (config::enable_abort_txn_and_job_for_delete_rowset_before_recycle)
{
- if (make_deferred_abort_task(rowset).has_value()) {
+ if (make_related_txn_or_job_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.tablet_id() << " version=["
<< rowset.start_version() << '-' <<
rowset.end_version() << "]";
tmp_rowset_keys_to_abort.emplace_back(k);
+ return 0;
Review Comment:
[P2] Preserve expired accounting when deferring tmp aborts
This new return skips `++num_expired` and `expired_rowset_size += v.size()`
below for every owner-associated tmp rowset. When the worker aborts and deletes
the key successfully it still increments `num_recycled`, so a run can report
`num_recycled > num_expired` and zero expired metadata bytes for the data it
just reclaimed. Record the expired key/value before deferring the abort (or
account from the worker's reread) and cover these completion fields in the
tmp-rowset tests.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -2081,77 +2147,125 @@ int collect_deferred_abort_tasks(TxnKv* txn_kv, const
std::string& instance_id,
template <typename T>
int InstanceRecycler::batch_abort_txn_or_job_for_recycle(const
std::vector<std::string>& keys,
- bool
skip_base_version) {
- std::vector<DeferredRecycleAbortTask> abort_tasks;
- if (collect_deferred_abort_tasks<T>(txn_kv_.get(), instance_id_, keys,
&abort_tasks,
- skip_base_version) != 0) {
+
std::vector<std::string>& aborted_keys) {
+ std::vector<RelatedTxnOrJobAbortTask> abort_tasks;
+ if (collect_deferred_abort_tasks<T>(txn_kv_.get(), instance_id_, keys,
&abort_tasks) != 0) {
LOG(WARNING) << "failed to collect rowset abort tasks, instance_id="
<< instance_id_;
return -1;
}
+ int ret = 0;
for (const auto& abort_task : abort_tasks) {
- LOG(INFO) << "begin to abort txn or job for related rowset,
instance_id=" << instance_id_
- << " tablet_id=" << abort_task.tablet_id << " version=["
- << abort_task.start_version << '-' << abort_task.end_version
<< "]";
int abort_ret = 0;
- if (abort_task.type == DeferredRecycleAbortTask::Type::TXN) {
+ if (abort_task.type == RelatedTxnOrJobAbortTask::Type::TXN) {
abort_ret = abort_txn_for_related_rowset(abort_task.txn_id);
} else {
- RowsetMetaCloudPB rowset_meta;
- rowset_meta.set_tablet_id(abort_task.tablet_id);
- rowset_meta.set_rowset_id_v2(abort_task.rowset_id);
- rowset_meta.set_job_id(abort_task.job_id);
- abort_ret = abort_job_for_related_rowset(rowset_meta);
+ abort_ret = abort_job_for_related_rowset(abort_task.tablet_id,
abort_task.rowset_id,
+ abort_task.job_id);
}
if (abort_ret != 0) {
LOG(WARNING) << "failed to abort txn or job for related rowset,
instance_id="
<< instance_id_ << " tablet_id=" <<
abort_task.tablet_id << " version=["
<< abort_task.start_version << '-' <<
abort_task.end_version << "]";
- return abort_ret;
+ ret = abort_ret;
+ continue;
}
+ aborted_keys.emplace_back(abort_task.key);
}
- return 0;
+ return ret;
}
-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;
+void InstanceRecycler::submit_recycle_tmp_rowsets_job(
+ SimpleThreadPool& worker_pool, std::vector<std::string>
tmp_rowset_keys_to_delete,
+ std::vector<std::string> rowset_keys_to_mark, std::vector<std::string>
rowset_keys_to_abort,
+ std::atomic_long* num_recycled, RecyclerMetricsContext*
metrics_context) {
+ if (tmp_rowset_keys_to_delete.empty() && rowset_keys_to_mark.empty() &&
+ rowset_keys_to_abort.empty()) {
+ return;
+ }
+ worker_pool.submit([this, tmp_rowset_keys_to_delete =
std::move(tmp_rowset_keys_to_delete),
+ rowset_keys_to_mark = std::move(rowset_keys_to_mark),
+ rowset_keys_to_abort =
std::move(rowset_keys_to_abort), num_recycled,
+ metrics_context]() mutable {
+ if (!rowset_keys_to_mark.empty() &&
+ batch_mark_rowsets_as_recycled<RowsetMetaCloudPB>(txn_kv_.get(),
instance_id_,
+
rowset_keys_to_mark) != 0) {
+ LOG(WARNING) << "failed to batch mark tmp rowsets as recycled,
instance_id="
+ << instance_id_;
+ return;
}
- for (size_t idx = offset; idx < limit; ++idx) {
- const std::string& key = keys[idx];
+
+ if (!rowset_keys_to_abort.empty() &&
+
batch_abort_txn_or_job_for_recycle<RowsetMetaCloudPB>(rowset_keys_to_abort,
+
tmp_rowset_keys_to_delete) != 0) {
+ LOG(WARNING) << "failed to abort some txn or job for related tmp
rowset, "
+ "instance_id="
+ << instance_id_;
+ }
+
+ // Keys that do not need an abort were added during the scan. Keys
related to a txn
+ // or job are added only after the abort succeeds. A failed abort
leaves its tmp
+ // rowset KV for the next recycler round.
+ std::map<std::string, RowsetMetaCloudPB> tmp_rowsets_to_delete;
+ std::vector<std::string> current_tmp_rowset_keys;
+ std::vector<std::string> current_tmp_rowset_ref_count_keys;
+ for (const auto& key : tmp_rowset_keys_to_delete) {
+ // The tmp rowset may have changed after the scan. Read it again
before deleting
+ // its object data and metadata.
std::string val;
- err = txn->get(key, &val);
- if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
- // has already been removed
+ int ret = txn_get(txn_kv_.get(), key, val);
Review Comment:
[P2] Batch the tmp-rowset freshness reads
This loop calls `txn_get()` once per key; that helper creates a new
transaction and synchronously waits for one snapshot point read. A scan page
can contain 10,000 expired tmp keys, so one worker can perform 10,000 serial
FDB round trips before any object cleanup begins. Keep the post-abort freshness
check, but use snapshot `batch_get` in bounded chunks (the existing API issues
up to 1,000 FDB gets concurrently) and retain the per-key missing/parse
handling. Please add a large-batch read-count or instrumentation 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]
