diff --git a/cloud/src/recycler/recycler.cpp b/cloud/src/recycler/recycler.cpp index bd1b9bff878cee..c590904fe4f189 100644 --- a/cloud/src/recycler/recycler.cpp +++ b/cloud/src/recycler/recycler.cpp @@ -1954,15 +1954,9 @@ struct DeferredRecyclePrepareDeleteTask { int64_t tablet_id = 0; }; -template -std::optional make_deferred_abort_task(const T& rowset_meta_pb) { - if constexpr (std::is_same_v) { - if (rowset_meta_pb.type() != RecycleRowsetPB::PREPARE) { - return std::nullopt; - } - } - - const auto& rs_meta = rowset_meta(rowset_meta_pb); +std::optional 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(); @@ -1981,10 +1975,8 @@ std::optional make_deferred_abort_task(const T& rowset return std::nullopt; } -template -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 @@ -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); @@ -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) { + 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)); } @@ -2112,48 +2109,6 @@ int InstanceRecycler::batch_abort_txn_or_job_for_recycle(const std::vector& keys, - std::vector* 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 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; @@ -5000,7 +4955,6 @@ int InstanceRecycler::recycle_rowsets() { }; std::vector rowset_keys_to_mark_recycled; - std::vector rowset_keys_to_abort; std::vector prepare_rowset_keys_to_delete; // Store keys of rowset recycled by background workers @@ -5010,9 +4964,6 @@ int InstanceRecycler::recycle_rowsets() { auto worker_pool = std::make_unique( config::instance_recycler_worker_pool_size, "recycle_rowsets"); worker_pool->start(); - auto mark_abort_worker_pool = std::make_unique( - 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 = @@ -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 @@ -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(), @@ -5221,21 +5177,17 @@ int InstanceRecycler::recycle_rowsets() { }); }; - auto submit_mark_abort_rowset_job = [&](std::vector rowset_keys_to_mark, - std::vector rowset_keys_to_abort) { - if (rowset_keys_to_mark.empty() && rowset_keys_to_abort.empty()) { + auto submit_mark_rowset_job = [&](std::vector 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(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(txn_kv_.get(), instance_id_, @@ -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(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 mark_keys_to_process; - std::vector 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; } @@ -5332,7 +5272,6 @@ int InstanceRecycler::recycle_rowsets() { ret = -1; } - mark_abort_worker_pool->stop(); worker_pool->stop(); if (!async_recycled_rowset_keys.empty()) { diff --git a/cloud/test/recycler_test.cpp b/cloud/test/recycler_test.cpp index ad8075cec9c55f..aadb39c5c95692 100644 --- a/cloud/test/recycler_test.cpp +++ b/cloud/test/recycler_test.cpp @@ -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);